diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9afcf0f3..534a4214 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -37,6 +37,7 @@ on: - feat/ar1-metadata-fabric-active-metadata-binding-reconciliation - feat/ar1-metadata-fabric-durable-active-metadata-promotion - feat/ar1-metadata-fabric-object-store-active-metadata-promotion + - feat/ar1-metadata-fabric-real-feature-ingestion env: PYTHON_VERSION: "3.13" @@ -200,6 +201,9 @@ jobs: - name: Validate metadata fabric real-feature ingestion evidence run: python -m data_agent.metadata_fabric_real_feature_ingestion validate + - name: Validate metadata fabric real-feature ledger promotion evidence + run: python -m data_agent.metadata_fabric_real_feature_ledger_promotion validate + - name: Validate Active Metadata consumer deployment boundary run: python -m data_agent.active_metadata_consumer_deployment validate @@ -244,10 +248,14 @@ jobs: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test run: python -m pytest data_agent/test_active_metadata_authorization_postgres.py -q - - name: Verify Active Metadata binding reconciliation on PostgreSQL + - name: Verify binding reconciliation and real-feature ledger promotion on PostgreSQL env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test - run: python -m pytest data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py -q + run: >- + python -m pytest + data_agent/test_metadata_fabric_active_metadata_binding_reconciliation_postgres.py + data_agent/test_metadata_fabric_real_feature_ledger_promotion_postgres.py + -q - name: Run required platform tests env: @@ -291,6 +299,7 @@ jobs: data_agent/test_metadata_fabric_durable_active_metadata_promotion.py \ data_agent/test_metadata_fabric_object_store_active_metadata_promotion.py \ data_agent/test_metadata_fabric_real_feature_ingestion.py \ + data_agent/test_metadata_fabric_real_feature_ledger_promotion.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_real_feature_ledger_promotion.py b/data_agent/metadata_fabric_real_feature_ledger_promotion.py new file mode 100644 index 00000000..9604d61c --- /dev/null +++ b/data_agent/metadata_fabric_real_feature_ledger_promotion.py @@ -0,0 +1,1266 @@ +"""Atomically promote the checked M3-22 real-feature output ledger bundle. + +M3-23 consumes only the path-free M3-22 evidence. It appends the output +ResourceVersion, output Artifact, quality evidence Artifact, independent passed +QualityResult, and source-to-output LineageEvent through one PlatformGateway +transaction. The correlated PlatformRun stays accepted: this slice does not +fabricate the success observation or independent evidence provenance required by +the existing terminal success gate. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import timedelta +from pathlib import Path +from typing import Any +from uuid import uuid5 + +from pydantic import BaseModel, ConfigDict, model_validator +from sqlalchemy import create_engine, text +from sqlalchemy.exc import DBAPIError + +from . import metadata_fabric_active_metadata_authorization as m316 +from . import metadata_fabric_real_feature_ingestion as m322 +from .platform_contracts import ( + Artifact, + LineageEvent, + PlatformDefinitionVersion, + PlatformRun, + QualityResult, + Resource, + ResourceVersion, + RunStatus, + RunSuccessEvidence, + SubjectContext, + canonical_json_fingerprint, + platform_definition_fingerprint, + run_success_evidence_fingerprint, +) +from .platform_gateway import ( + DefinitionRegistration, + GatewayConflictError, + GatewayNotFoundError, + GatewayValidationError, + GatewayWriteResult, + PlatformGateway, + PlatformGatewayError, +) + +CONTRACT_SCHEMA = "gda.real_feature_ledger_promotion_contract.v1" +EVIDENCE_SCHEMA = "gda.real_feature_ledger_promotion_evidence.v1" +VALIDATION_SCHEMA = "gda.real_feature_ledger_promotion_validation.v1" +SOURCE_EVIDENCE_SHA256 = ( + "42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899" +) +SOURCE_CONTRACT_SHA256 = ( + "af211f2d2f4830decb9ffe369cd9e7ec2c9349c2e2c8bd789347a6fdc288e1dc" +) +TENANT = m322.TENANT +SOURCE_RESOURCE_VERSION_ID = m322.SOURCE_RESOURCE_VERSION_ID +OUTPUT_RESOURCE_VERSION_ID = m322.OUTPUT_RESOURCE_VERSION_ID +DEFINITION_VERSION_ID = m322.DEFINITION_VERSION_ID +RUN_ID = m322.RUN_ID +WORKLOAD = "workload:real-feature-ledger-promoter" +REPO_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_SOURCE_EVIDENCE_PATH = m322.DEFAULT_EVIDENCE_PATH +DEFAULT_EVIDENCE_PATH = ( + REPO_ROOT + / "docs/evidence/metadata-fabric-real-feature-ledger-promotion-2026-07-31.json" +) +DEFAULT_WRAPPER_PATH = ( + REPO_ROOT / "scripts/metadata-fabric-real-feature-ledger-promotion.sh" +) +MIGRATIONS = tuple( + Path(__file__).resolve().parent / "migrations" / filename + for filename in ( + "092_platform_control_ledger.sql", + "093_app_user_tenant_context.sql", + "094_platform_control_gateway.sql", + "095_platform_command_outbox.sql", + "096_platform_success_verdict.sql", + ) +) +FALSE_CLAIMS = ( + "source_dataset_committed", + "source_absolute_path_committed", + "source_feature_payload_committed", + "m322_authorization_persisted_to_gda_control", + "output_material_retained", + "platform_run_succeeded", + "protected_workload_identity_verified", + "durable_catalog_verified", + "production_object_store_verified", + "production_ingestion_verified", + "production_ready", +) + + +class RealFeatureLedgerPromotionError(RuntimeError): + """The M3-23 output ledger promotion failed closed.""" + + +class _InjectedPromotionFailure(RuntimeError): + pass + + +class RunOutputLedgerPromotion(BaseModel): + """Atomic, content-bound output, quality and lineage bundle.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + authority_resource: Resource + output_resource_version: ResourceVersion + output_artifact: Artifact + quality_evidence_artifact: Artifact + quality_result: QualityResult + lineage_event: LineageEvent + + @model_validator(mode="after") + def _consistent_promotion(self) -> RunOutputLedgerPromotion: + resource = self.authority_resource + version = self.output_resource_version + output = self.output_artifact + quality_artifact = self.quality_evidence_artifact + quality = self.quality_result + lineage = self.lineage_event + if len( + { + resource.tenant_id, + version.tenant_id, + output.tenant_id, + quality_artifact.tenant_id, + quality.tenant_id, + lineage.tenant_id, + } + ) != 1: + raise ValueError("output promotion tenants must match") + if resource.resource_urn != version.resource_urn: + raise ValueError("output ResourceVersion must bind the authority Resource") + if output.artifact_role.value != "output": + raise ValueError("output promotion requires one output Artifact") + if quality_artifact.artifact_role.value != "evidence": + raise ValueError("output promotion requires one quality evidence Artifact") + if output.artifact_id == quality_artifact.artifact_id: + raise ValueError("output and quality evidence Artifacts must be distinct") + if ( + output.run_id is None + or quality_artifact.run_id != output.run_id + or quality.run_id != output.run_id + or lineage.run_id != output.run_id + ): + raise ValueError("output promotion Run bindings must match") + if ( + output.resource_version_id != version.resource_version_id + or quality_artifact.resource_version_id != version.resource_version_id + or quality.resource_version_id != version.resource_version_id + or lineage.target_resource_version_id != version.resource_version_id + ): + raise ValueError("output promotion ResourceVersion bindings must match") + if output.content_sha256 != version.content_sha256: + raise ValueError("output Artifact must bind ResourceVersion content") + if quality.evidence_artifact_id != quality_artifact.artifact_id: + raise ValueError("QualityResult must bind the quality evidence Artifact") + if quality.verdict.value != "passed": + raise ValueError("output promotion requires a passed QualityResult") + if quality.evaluated_by == version.created_by: + raise ValueError("output promotion quality evaluation must be independent") + if quality_artifact.manifest.get("rule_version_ref") != quality.rule_version_ref: + raise ValueError("quality evidence rule binding does not match") + if quality_artifact.manifest.get("metrics") != quality.metrics: + raise ValueError("quality evidence metrics do not match") + if lineage.artifact_id != output.artifact_id: + raise ValueError("LineageEvent must bind the output Artifact") + if lineage.definition_version_id is None: + raise ValueError("output promotion lineage requires a DefinitionVersion") + if any( + output.manifest.get(key) != value + for key, value in lineage.facets.items() + ): + raise ValueError("lineage facets do not match the output manifest") + return self + + +class RunOutputLedgerPromoter: + """Compose existing gateway primitives under one gateway transaction.""" + + def __init__(self, gateway: PlatformGateway): + self.gateway = gateway + + @staticmethod + def _sql_json(value: Any) -> str: + return json.dumps( + value, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ) + + def _put_artifact(self, connection, artifact: Artifact) -> GatewayWriteResult: + inserted = connection.execute( + text( + """ + INSERT INTO gda_control.artifact ( + tenant_id, artifact_id, artifact_key, artifact_role, + storage_uri, media_type, content_sha256, size_bytes, + run_id, resource_version_id, manifest, created_by, created_at + ) VALUES ( + :tenant_id, :artifact_id, :artifact_key, :artifact_role, + :storage_uri, :media_type, :content_sha256, :size_bytes, + :run_id, :resource_version_id, + CAST(:manifest AS jsonb), :created_by, :created_at + ) + ON CONFLICT DO NOTHING + RETURNING artifact_id + """ + ), + { + **artifact.model_dump(mode="python", exclude={"manifest"}), + "artifact_role": artifact.artifact_role.value, + "manifest": self._sql_json(artifact.manifest), + }, + ).first() + stored = self.gateway._load_artifact( + connection, artifact.tenant_id, artifact.artifact_id + ) + if stored is None or stored != artifact: + raise GatewayConflictError( + "Artifact identity already has a different payload" + ) + return GatewayWriteResult(stored, inserted is not None) + + def _put_quality_result( + self, connection, quality: QualityResult + ) -> GatewayWriteResult: + inserted = connection.execute( + text( + """ + INSERT INTO gda_control.quality_result ( + tenant_id, quality_result_id, run_id, + resource_version_id, rule_version_ref, verdict, + metrics, evidence_artifact_id, result_sha256, + evaluated_by, evaluated_at + ) VALUES ( + :tenant_id, :quality_result_id, :run_id, + :resource_version_id, :rule_version_ref, :verdict, + CAST(:metrics AS jsonb), :evidence_artifact_id, + :result_sha256, :evaluated_by, :evaluated_at + ) + ON CONFLICT DO NOTHING + RETURNING quality_result_id + """ + ), + { + **quality.model_dump(mode="python", exclude={"metrics"}), + "verdict": quality.verdict.value, + "metrics": self._sql_json(quality.metrics), + }, + ).first() + stored = self.gateway._load_quality_result( + connection, + quality.tenant_id, + quality.quality_result_id, + ) + if stored is None or stored != quality: + raise GatewayConflictError( + "QualityResult identity already has a different payload" + ) + return GatewayWriteResult(stored, inserted is not None) + + def _put_lineage( + self, connection, event: LineageEvent + ) -> GatewayWriteResult: + inserted = connection.execute( + text( + """ + INSERT INTO gda_control.lineage_event ( + tenant_id, lineage_event_id, event_type, + source_resource_version_id, target_resource_version_id, + producer, event_sha256, run_id, definition_version_id, + artifact_id, facets, occurred_at + ) VALUES ( + :tenant_id, :lineage_event_id, :event_type, + :source_resource_version_id, :target_resource_version_id, + :producer, :event_sha256, :run_id, :definition_version_id, + :artifact_id, CAST(:facets AS jsonb), :occurred_at + ) + ON CONFLICT DO NOTHING + RETURNING lineage_event_id + """ + ), + { + **event.model_dump(mode="python", exclude={"facets"}), + "event_type": event.event_type.value, + "facets": self._sql_json(event.facets), + }, + ).first() + stored = self.gateway._load_lineage( + connection, event.tenant_id, event.lineage_event_id + ) + if stored is None or stored != event: + raise GatewayConflictError( + "LineageEvent identity already has a different payload" + ) + return GatewayWriteResult(stored, inserted is not None) + + def promote( + self, promotion: RunOutputLedgerPromotion + ) -> GatewayWriteResult: + try: + promotion = RunOutputLedgerPromotion.model_validate( + promotion.model_dump(mode="json") + ) + except ValueError as exc: + raise GatewayValidationError( + "Run output ledger promotion is not content-bound" + ) from exc + tenant_id = promotion.output_resource_version.tenant_id + run_id = promotion.output_artifact.run_id + assert run_id is not None + gateway = self.gateway + with gateway._transaction(tenant_id) as connection: + authority = gateway._load_resource( + connection, + tenant_id, + promotion.output_resource_version.resource_urn, + ) + if authority is None: + raise GatewayValidationError( + "output Resource authority record was not found" + ) + if authority != promotion.authority_resource: + raise GatewayConflictError( + "output Resource authority record has different content" + ) + run = gateway._load_run(connection, tenant_id, run_id) + source = gateway._load_resource_version( + connection, + tenant_id, + promotion.lineage_event.source_resource_version_id, + ) + definition = gateway._load_definition( + connection, + tenant_id, + promotion.lineage_event.definition_version_id, + ) + if run is None or source is None or definition is None: + raise GatewayValidationError( + "Run output promotion prerequisite was not found" + ) + if ( + run.definition_version_id != definition.definition_version_id + or run.definition_version_id + != promotion.lineage_event.definition_version_id + or promotion.lineage_event.source_resource_version_id + not in { + binding.resource_version_id for binding in run.input_bindings + } + ): + raise GatewayValidationError( + "Run output promotion prerequisite binding does not match" + ) + if run.status not in {RunStatus.ACCEPTED, RunStatus.RECONCILING}: + raise GatewayValidationError( + "Run output promotion requires an accepted or reconciling Run" + ) + + results = ( + gateway._put_resource_version( + connection, promotion.output_resource_version + ), + self._put_artifact(connection, promotion.output_artifact), + self._put_artifact( + connection, promotion.quality_evidence_artifact + ), + self._put_quality_result(connection, promotion.quality_result), + self._put_lineage(connection, promotion.lineage_event), + ) + creation_states = {result.created for result in results} + if len(creation_states) != 1: + raise GatewayConflictError( + "Run output promotion has partial pre-existing state" + ) + final_run = gateway._load_run(connection, tenant_id, run_id) + if final_run != run: + raise GatewayConflictError( + "Run changed while its output ledger was promoted" + ) + stored = RunOutputLedgerPromotion( + authority_resource=authority, + output_resource_version=results[0].value, + output_artifact=results[1].value, + quality_evidence_artifact=results[2].value, + quality_result=results[3].value, + lineage_event=results[4].value, + ) + return GatewayWriteResult(stored, results[0].created) + + +class _RollbackProbePromoter(RunOutputLedgerPromoter): + def _put_quality_result(self, connection, quality): + raise _InjectedPromotionFailure("injected before QualityResult append") + + +@dataclass(frozen=True) +class PromotionPrerequisites: + source_resource: Resource + source_version: ResourceVersion + definition_registration: DefinitionRegistration + output_resource: Resource + run: PlatformRun + + +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 RealFeatureLedgerPromotionError( + f"{path.name} is not valid JSON" + ) from exc + if not isinstance(value, dict): + raise RealFeatureLedgerPromotionError(f"{path.name} must contain an object") + return value + + +def _mapping(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, Mapping) else {} + + +def _file_record(path: Path) -> dict[str, str | None]: + resolved = path.resolve() + return { + "path": resolved.relative_to(REPO_ROOT).as_posix(), + "sha256": ( + hashlib.sha256(resolved.read_bytes()).hexdigest() + if resolved.is_file() + else None + ), + } + + +def validate_source_evidence(source: Mapping[str, Any]) -> None: + errors = m322.verify_evidence_integrity(source) + if errors: + raise RealFeatureLedgerPromotionError( + "M3-22 evidence is invalid: " + ", ".join(errors) + ) + if ( + source.get("evidence_sha256") != SOURCE_EVIDENCE_SHA256 + or source.get("contract_sha256") != SOURCE_CONTRACT_SHA256 + ): + raise RealFeatureLedgerPromotionError("M3-22 evidence identity drifted") + if ( + source.get("ingestion_persisted_to_gda_control") is not False + or source.get("platform_run_succeeded") is not False + ): + raise RealFeatureLedgerPromotionError( + "M3-22 source must contain unpromoted, non-terminal candidates" + ) + + +def build_output_authority_resource(source: Mapping[str, Any]) -> Resource: + validate_source_evidence(source) + observation = _mapping(source.get("observation")) + plan = _mapping(observation.get("plan")) + target = _mapping(plan.get("target")) + contracts = _mapping(observation.get("output_contracts")) + output = ResourceVersion.model_validate(contracts.get("output_resource_version")) + artifact = Artifact.model_validate(contracts.get("output_artifact")) + locator = "/".join( + str(target.get(name) or "") + for name in ("metalake", "catalog", "schema", "table") + ) + if not all(target.get(name) for name in ("metalake", "catalog", "schema", "table")): + raise RealFeatureLedgerPromotionError("M3-22 authority target is incomplete") + return Resource( + tenant_id=TENANT, + resource_urn=output.resource_urn, + resource_kind="data_product", + authority_system="gravitino", + authority_locator=locator, + owner_ref="team:metadata-platform", + governance_ref={ + "claim_level": "local_verified_ingestion_output", + "source_evidence_schema": m322.EVIDENCE_SCHEMA, + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "production_ready": False, + }, + technical_refs=( + { + "system": "iceberg", + "catalog": target["catalog"], + "schema": target["schema"], + "table": target["table"], + "content_sha256": output.content_sha256, + }, + { + "system": "object_store", + "storage_uri": artifact.storage_uri, + "material_retained": False, + }, + ), + ) + + +def build_promotion(source: Mapping[str, Any]) -> RunOutputLedgerPromotion: + validate_source_evidence(source) + contracts = _mapping( + _mapping(source.get("observation")).get("output_contracts") + ) + if contracts.get("persisted_to_gda_control") is not False: + raise RealFeatureLedgerPromotionError("M3-22 candidates already claim persistence") + try: + promotion = RunOutputLedgerPromotion( + authority_resource=build_output_authority_resource(source), + output_resource_version=ResourceVersion.model_validate( + contracts.get("output_resource_version") + ), + output_artifact=Artifact.model_validate(contracts.get("output_artifact")), + quality_evidence_artifact=Artifact.model_validate( + contracts.get("quality_evidence_artifact") + ), + quality_result=contracts.get("quality_result"), + lineage_event=contracts.get("lineage_event"), + ) + except ValueError as exc: + raise RealFeatureLedgerPromotionError( + "M3-22 output ledger candidates are invalid" + ) from exc + plan = _mapping(_mapping(source.get("observation")).get("plan")) + if ( + promotion.output_resource_version.resource_version_id + != OUTPUT_RESOURCE_VERSION_ID + or promotion.output_resource_version.resource_urn != m322.OUTPUT_RESOURCE_URN + or promotion.lineage_event.source_resource_version_id + != SOURCE_RESOURCE_VERSION_ID + or promotion.lineage_event.definition_version_id != DEFINITION_VERSION_ID + or promotion.output_artifact.run_id != RUN_ID + or promotion.output_resource_version.content_sha256 + != plan.get("output_content_sha256") + ): + raise RealFeatureLedgerPromotionError("M3-22 promotion identity drifted") + return promotion + + +def build_prerequisites( + source: Mapping[str, Any], promotion: RunOutputLedgerPromotion +) -> PromotionPrerequisites: + validate_source_evidence(source) + observation = _mapping(source.get("observation")) + plan = _mapping(observation.get("plan")) + authorization = _mapping(observation.get("authorization")) + source_bundle = m316.build_authorization_bundle( + str(plan.get("source_content_sha256") or "") + ) + source_resource = source_bundle.source_resource + source_version = source_bundle.registration.resource_version + if ( + source_version.resource_version_id != SOURCE_RESOURCE_VERSION_ID + or source_version.content_sha256 != plan.get("source_content_sha256") + ): + raise RealFeatureLedgerPromotionError("M3-22 source prerequisite drifted") + + definition_urn = f"gda://{TENANT}/definition/real-feature-ingestion" + definition_document = { + "action": m322.ACTION, + "ingestion_plan_sha256": plan.get("ingestion_plan_sha256"), + "engine": "spark-sedona-iceberg", + "terminal_success": False, + } + input_contract = { + "source_resource_version_id": str(SOURCE_RESOURCE_VERSION_ID), + "semantic_type": "gis.cultural_districts", + } + output_contract = { + "output_resource_version_id": str(OUTPUT_RESOURCE_VERSION_ID), + "output_artifact": True, + "independent_quality_result": True, + "source_to_output_lineage": True, + "platform_run_terminal_success": False, + } + definition_sha256 = platform_definition_fingerprint( + orchestration_class="dataops", + capability_id=m322.ACTION, + portability_class="engine_family", + definition_document=definition_document, + input_contract=input_contract, + output_contract=output_contract, + ) + created_at = promotion.output_resource_version.created_at - timedelta(minutes=5) + definition_resource = Resource( + tenant_id=TENANT, + resource_urn=definition_urn, + resource_kind="definition", + authority_system="gda", + authority_locator="definition/real-feature-ingestion", + owner_ref="team:metadata-platform", + ) + definition_version = ResourceVersion( + tenant_id=TENANT, + resource_urn=definition_urn, + resource_version_id=DEFINITION_VERSION_ID, + version_key="m3-22-spark-sedona-iceberg-v1", + content_sha256=definition_sha256, + authority_version_ref={ + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "ingestion_plan_sha256": plan.get("ingestion_plan_sha256"), + }, + created_by=WORKLOAD, + created_at=created_at, + ) + definition = PlatformDefinitionVersion( + tenant_id=TENANT, + definition_urn=definition_urn, + definition_version_id=DEFINITION_VERSION_ID, + orchestration_class="dataops", + capability_id=m322.ACTION, + portability_class="engine_family", + definition_document=definition_document, + input_contract=input_contract, + output_contract=output_contract, + definition_sha256=definition_sha256, + ) + run = PlatformRun( + tenant_id=TENANT, + run_id=RUN_ID, + definition_version_id=DEFINITION_VERSION_ID, + orchestration_class="dataops", + subject_context=SubjectContext( + tenant_id=TENANT, + subject_id=m322.WORKLOAD.removeprefix("workload:"), + subject_type="workload", + roles=("spatial_ingestion_executor",), + purpose="correlate checked M3-22 output candidates for ledger promotion", + ), + input_bindings=( + { + "binding_name": "source_dataset", + "resource_version_id": SOURCE_RESOURCE_VERSION_ID, + "semantic_type": "gis.cultural_districts", + }, + ), + idempotency_key=f"real-feature-ingestion:{promotion.output_resource_version.content_sha256}", + config_fingerprint=str(authorization.get("authorization_sha256") or ""), + submitted_at=created_at + timedelta(seconds=1), + ) + return PromotionPrerequisites( + source_resource=source_resource, + source_version=source_version, + definition_registration=DefinitionRegistration( + resource=definition_resource, + resource_version=definition_version, + definition=definition, + ), + output_resource=promotion.authority_resource, + run=run, + ) + + +def _apply_migrations(engine: Any) -> None: + with engine.begin() as connection: + connection.exec_driver_sql( + """ + CREATE TABLE IF NOT EXISTS agent_app_users ( + id SERIAL PRIMARY KEY, + username VARCHAR(100) UNIQUE NOT NULL + ) + """ + ) + for migration in MIGRATIONS: + connection.execute(text(migration.read_text(encoding="utf-8"))) + + +def _register_prerequisites( + gateway: PlatformGateway, + prerequisites: PromotionPrerequisites, + *, + include_output_authority: bool, +) -> None: + gateway.register_resource(prerequisites.source_resource) + gateway.register_resource_version(prerequisites.source_version) + gateway.register_definition(prerequisites.definition_registration) + if include_output_authority: + gateway.register_resource(prerequisites.output_resource) + gateway.submit_run(prerequisites.run) + + +def _candidate_counts(engine: Any, promotion: RunOutputLedgerPromotion) -> dict[str, int]: + with engine.connect() as connection: + return { + "resource_versions": int( + connection.execute( + text( + "SELECT count(*) FROM gda_control.resource_version " + "WHERE tenant_id=:tenant_id AND resource_version_id=:object_id" + ), + { + "tenant_id": TENANT, + "object_id": promotion.output_resource_version.resource_version_id, + }, + ).scalar_one() + ), + "artifacts": int( + connection.execute( + text( + "SELECT count(*) FROM gda_control.artifact " + "WHERE tenant_id=:tenant_id AND artifact_id IN (:output_id, :quality_id)" + ), + { + "tenant_id": TENANT, + "output_id": promotion.output_artifact.artifact_id, + "quality_id": promotion.quality_evidence_artifact.artifact_id, + }, + ).scalar_one() + ), + "quality_results": int( + connection.execute( + text( + "SELECT count(*) FROM gda_control.quality_result " + "WHERE tenant_id=:tenant_id AND quality_result_id=:object_id" + ), + { + "tenant_id": TENANT, + "object_id": promotion.quality_result.quality_result_id, + }, + ).scalar_one() + ), + "lineage_events": int( + connection.execute( + text( + "SELECT count(*) FROM gda_control.lineage_event " + "WHERE tenant_id=:tenant_id AND lineage_event_id=:object_id" + ), + { + "tenant_id": TENANT, + "object_id": promotion.lineage_event.lineage_event_id, + }, + ).scalar_one() + ), + } + + +def _security_state(engine: Any) -> dict[str, bool]: + relations = ( + "resource_version", + "artifact", + "quality_result", + "lineage_event", + ) + with engine.connect() as connection: + force_rls = connection.execute( + text( + """ + SELECT bool_and(relforcerowsecurity) + FROM pg_class + WHERE oid = ANY (ARRAY[ + 'gda_control.resource_version'::regclass, + 'gda_control.artifact'::regclass, + 'gda_control.quality_result'::regclass, + 'gda_control.lineage_event'::regclass + ]) + """ + ) + ).scalar_one() + privileges = [ + connection.execute( + text( + """ + SELECT + has_table_privilege('gda_control_gateway', :relation, 'SELECT,INSERT'), + NOT has_table_privilege('gda_control_gateway', :relation, 'UPDATE'), + NOT has_table_privilege('gda_control_gateway', :relation, 'DELETE') + """ + ), + {"relation": f"gda_control.{name}"}, + ).one() + for name in relations + ] + return { + "force_rls": bool(force_rls), + "minimum_grants": all(bool(row[0] and row[1] and row[2]) for row in privileges), + } + + +def _direct_mutations_blocked( + gateway: PlatformGateway, promotion: RunOutputLedgerPromotion +) -> dict[str, bool]: + targets = { + "resource_version": ( + "resource_version_id", + promotion.output_resource_version.resource_version_id, + ), + "artifact": ("artifact_id", promotion.output_artifact.artifact_id), + "quality_result": ( + "quality_result_id", + promotion.quality_result.quality_result_id, + ), + "lineage_event": ( + "lineage_event_id", + promotion.lineage_event.lineage_event_id, + ), + } + results: dict[str, bool] = {} + with gateway._transaction(TENANT) as connection: + for relation, (identity_column, identity) in targets.items(): + for operation in ("update", "delete"): + statement = ( + f"UPDATE gda_control.{relation} SET tenant_id=tenant_id " + f"WHERE {identity_column}=:identity" + if operation == "update" + else f"DELETE FROM gda_control.{relation} " + f"WHERE {identity_column}=:identity" + ) + blocked = False + try: + with connection.begin_nested(): + connection.execute(text(statement), {"identity": identity}) + except DBAPIError: + blocked = True + results[f"{relation}_{operation}"] = blocked + return results + + +def _cross_tenant_direct_insert_blocked(gateway: PlatformGateway) -> bool: + blocked = False + with gateway._transaction(TENANT) as connection: + try: + with connection.begin_nested(): + connection.execute( + text( + """ + INSERT INTO gda_control.resource ( + tenant_id, resource_urn, resource_kind, + authority_system, authority_locator, owner_ref + ) VALUES ( + 'isolated-tenant', + 'gda://isolated-tenant/data_product/forbidden-output', + 'data_product', 'gda', 'forbidden-output', 'team:test' + ) + """ + ) + ) + except DBAPIError: + blocked = True + return blocked + + +def _success_finalization_rejected( + gateway: PlatformGateway, promotion: RunOutputLedgerPromotion +) -> bool: + observation_id = uuid5(RUN_ID, "missing-m3-23-success-observation") + values = { + "tenant_id": TENANT, + "run_id": RUN_ID, + "attempt_observation_id": observation_id, + "output_artifact_id": promotion.output_artifact.artifact_id, + "quality_result_id": promotion.quality_result.quality_result_id, + "lineage_event_id": promotion.lineage_event.lineage_event_id, + } + evidence = RunSuccessEvidence( + **values, + evidence_sha256=run_success_evidence_fingerprint(**values), + ) + try: + gateway.finalize_run_success( + evidence, + expected_state_version=0, + actor_subject=m322.WORKLOAD, + reason="must remain rejected until terminal evidence is complete", + ) + except PlatformGatewayError: + return True + return False + + +def run_postgres_rehearsal( + database_url: str, + *, + source_evidence_path: Path = DEFAULT_SOURCE_EVIDENCE_PATH, +) -> dict[str, Any]: + source = _load_json_object(source_evidence_path) + promotion = build_promotion(source) + prerequisites = build_prerequisites(source, promotion) + contract = build_contract_report(source_evidence_path=source_evidence_path) + if contract.get("status") != "valid": + raise RealFeatureLedgerPromotionError("M3-23 static contract is invalid") + engine = create_engine(database_url) + try: + _apply_migrations(engine) + gateway = PlatformGateway(engine) + promoter = RunOutputLedgerPromoter(gateway) + _register_prerequisites( + gateway, + prerequisites, + include_output_authority=False, + ) + missing_authority_rejected = False + try: + promoter.promote(promotion) + except GatewayValidationError: + missing_authority_rejected = True + gateway.register_resource(prerequisites.output_resource) + + rollback_injected = False + try: + _RollbackProbePromoter(gateway).promote(promotion) + except _InjectedPromotionFailure: + rollback_injected = True + rollback_counts = _candidate_counts(engine, promotion) + + first = promoter.promote(promotion) + replay = promoter.promote(promotion) + counts = _candidate_counts(engine, promotion) + run_before_finalization = gateway.get_run(TENANT, RUN_ID) + success_finalization_rejected = _success_finalization_rejected( + gateway, promotion + ) + final_run = gateway.get_run(TENANT, RUN_ID) + cross_tenant_read_blocked = False + try: + gateway.get_artifact( + "isolated-tenant", promotion.output_artifact.artifact_id + ) + except GatewayNotFoundError: + cross_tenant_read_blocked = True + direct_mutations = _direct_mutations_blocked(gateway, promotion) + cross_tenant_direct_insert_blocked = _cross_tenant_direct_insert_blocked( + gateway + ) + security = _security_state(engine) + with engine.connect() as connection: + run_event_count = int( + connection.execute( + text( + "SELECT count(*) FROM gda_control.platform_run_event " + "WHERE tenant_id=:tenant_id AND run_id=:run_id" + ), + {"tenant_id": TENANT, "run_id": RUN_ID}, + ).scalar_one() + ) + verified = ( + missing_authority_rejected + and rollback_injected + and rollback_counts + == { + "resource_versions": 0, + "artifacts": 0, + "quality_results": 0, + "lineage_events": 0, + } + and first.created + and not replay.created + and first.value == replay.value == promotion + and counts + == { + "resource_versions": 1, + "artifacts": 2, + "quality_results": 1, + "lineage_events": 1, + } + and run_before_finalization == final_run + and final_run.status == RunStatus.ACCEPTED + and final_run.state_version == 0 + and run_event_count == 1 + and success_finalization_rejected + and cross_tenant_read_blocked + and cross_tenant_direct_insert_blocked + and all(direct_mutations.values()) + and security["force_rls"] + and security["minimum_grants"] + ) + stable = { + "schema": EVIDENCE_SCHEMA, + "status": ( + "local_real_feature_ledger_promotion_verified" + if verified + else "blocked" + ), + "contract_sha256": contract["contract_sha256"], + "source_evidence_schema": m322.EVIDENCE_SCHEMA, + "source_evidence_sha256": SOURCE_EVIDENCE_SHA256, + "promotion_sha256": canonical_json_fingerprint( + promotion.model_dump(mode="json") + ), + "tenant_id": TENANT, + "source_resource_version_id": str(SOURCE_RESOURCE_VERSION_ID), + "output_resource_urn": promotion.output_resource_version.resource_urn, + "output_resource_version_id": str(OUTPUT_RESOURCE_VERSION_ID), + "output_content_sha256": promotion.output_resource_version.content_sha256, + "run_id": str(RUN_ID), + "definition_version_id": str(DEFINITION_VERSION_ID), + "authority_system": promotion.authority_resource.authority_system, + "authority_locator": promotion.authority_resource.authority_locator, + "missing_authority_rejected": missing_authority_rejected, + "failure_injection_rollback_verified": ( + rollback_injected and not any(rollback_counts.values()) + ), + "rollback_candidate_counts": rollback_counts, + "first_promotion_created": first.created, + "replay_promotion_created": replay.created, + "candidate_row_counts": counts, + "exact_replay_verified": first.value == replay.value == promotion, + "cross_tenant_read_blocked": cross_tenant_read_blocked, + "cross_tenant_direct_insert_blocked": ( + cross_tenant_direct_insert_blocked + ), + "direct_mutations_blocked": direct_mutations, + "force_rls_verified": security["force_rls"], + "minimum_grants_verified": security["minimum_grants"], + "platform_run_status": final_run.status.value, + "platform_run_state_version": final_run.state_version, + "platform_run_event_count": run_event_count, + "success_finalization_rejected": success_finalization_rejected, + "promotion_persisted_to_gda_control": verified, + "writes_to_legacy": False, + **{claim: False for claim in FALSE_CLAIMS}, + "errors": [] if verified else ["M3-23 PostgreSQL rehearsal did not verify"], + } + return {**stable, "evidence_sha256": canonical_json_fingerprint(stable)} + finally: + engine.dispose() + + +def build_contract_report( + *, source_evidence_path: Path = DEFAULT_SOURCE_EVIDENCE_PATH +) -> dict[str, Any]: + errors: list[str] = [] + source: dict[str, Any] | None = None + promotion: RunOutputLedgerPromotion | None = None + try: + source = _load_json_object(source_evidence_path) + promotion = build_promotion(source) + build_prerequisites(source, promotion) + except (OSError, TypeError, ValueError, RealFeatureLedgerPromotionError) as exc: + errors.append(f"M3-23 source contract is invalid: {type(exc).__name__}") + paths = ( + Path(__file__).resolve(), + (REPO_ROOT / "data_agent/platform_gateway.py").resolve(), + DEFAULT_WRAPPER_PATH.resolve(), + *(migration.resolve() for migration in MIGRATIONS), + ) + files = [_file_record(path) for path in paths] + if any(item["sha256"] is None for item in files): + errors.append("M3-23 contract file is missing") + promotion_source = Path(__file__).read_text(encoding="utf-8") + for marker in ( + "class RunOutputLedgerPromotion", + "class RunOutputLedgerPromoter", + "def promote(", + "Run output promotion has partial pre-existing state", + "Run output promotion requires an accepted or reconciling Run", + ): + if marker not in promotion_source: + errors.append(f"M3-23 promoter is missing marker: {marker}") + stable = { + "schema": CONTRACT_SCHEMA, + "status": "valid" if not errors else "invalid", + "source_evidence_sha256": ( + source.get("evidence_sha256") if source is not None else None + ), + "source_contract_sha256": ( + source.get("contract_sha256") if source is not None else None + ), + "promotion_sha256": ( + canonical_json_fingerprint(promotion.model_dump(mode="json")) + if promotion is not None + else None + ), + "atomic_write_order": [ + "resource_version", + "output_artifact", + "quality_evidence_artifact", + "quality_result", + "lineage_event", + ], + "requires_preexisting_output_authority": True, + "partial_preexisting_state_rejected": True, + "exact_replay_idempotent": True, + "platform_run_terminal_success": False, + "writes_to_legacy": False, + "files": files, + **{claim: False for claim in FALSE_CLAIMS}, + "errors": errors, + } + return {**stable, "contract_sha256": canonical_json_fingerprint(stable)} + + +def validate_rehearsal_evidence( + evidence: Mapping[str, Any], + *, + source_evidence_path: Path = DEFAULT_SOURCE_EVIDENCE_PATH, +) -> list[str]: + errors: list[str] = [] + try: + source = _load_json_object(source_evidence_path) + validate_source_evidence(source) + except RealFeatureLedgerPromotionError as exc: + return [str(exc)] + contract = build_contract_report(source_evidence_path=source_evidence_path) + stable = {key: value for key, value in evidence.items() if key != "evidence_sha256"} + if evidence.get("evidence_sha256") != canonical_json_fingerprint(stable): + errors.append("M3-23 evidence SHA-256 does not match") + if evidence.get("schema") != EVIDENCE_SCHEMA or evidence.get("errors") != []: + errors.append("M3-23 evidence is not verified") + if evidence.get("contract_sha256") != contract.get("contract_sha256"): + errors.append("M3-23 evidence contract SHA drifted") + if evidence.get("source_evidence_sha256") != SOURCE_EVIDENCE_SHA256: + errors.append("M3-23 evidence source SHA drifted") + expected_true = ( + "missing_authority_rejected", + "failure_injection_rollback_verified", + "first_promotion_created", + "exact_replay_verified", + "cross_tenant_read_blocked", + "cross_tenant_direct_insert_blocked", + "force_rls_verified", + "minimum_grants_verified", + "success_finalization_rejected", + "promotion_persisted_to_gda_control", + ) + for claim in expected_true: + if evidence.get(claim) is not True: + errors.append(f"M3-23 evidence claim is false: {claim}") + if evidence.get("replay_promotion_created") is not False: + errors.append("M3-23 replay must not create rows") + if evidence.get("candidate_row_counts") != { + "resource_versions": 1, + "artifacts": 2, + "quality_results": 1, + "lineage_events": 1, + }: + errors.append("M3-23 candidate row counts do not match") + if evidence.get("platform_run_status") != "accepted": + errors.append("M3-23 PlatformRun must remain accepted") + if evidence.get("platform_run_state_version") != 0: + errors.append("M3-23 PlatformRun state version changed") + direct_mutations = _mapping(evidence.get("direct_mutations_blocked")) + if len(direct_mutations) != 8 or not all(direct_mutations.values()): + errors.append("M3-23 direct mutation rejection is incomplete") + for claim in FALSE_CLAIMS: + if evidence.get(claim) is not False: + errors.append(f"M3-23 evidence may not claim {claim}") + if evidence.get("writes_to_legacy") is not False: + errors.append("M3-23 evidence may not claim legacy writes") + serialized = json.dumps(evidence, ensure_ascii=True, sort_keys=True) + for forbidden in ( + "/Users/", + "/home/", + "Downloads/", + ".tmp/", + "geometry_wkb_hex", + '"rows"', + '"password"', + '"secret"', + '"token"', + '"access_key"', + ): + if forbidden in serialized: + errors.append("M3-23 evidence contains source or secret material") + break + return errors + + +def build_validation_report( + *, + source_evidence_path: Path = DEFAULT_SOURCE_EVIDENCE_PATH, + evidence_path: Path = DEFAULT_EVIDENCE_PATH, +) -> dict[str, Any]: + contract = build_contract_report(source_evidence_path=source_evidence_path) + errors = list(contract["errors"]) + evidence: dict[str, Any] | None = None + try: + evidence = _load_json_object(evidence_path) + errors.extend( + validate_rehearsal_evidence( + evidence, + source_evidence_path=source_evidence_path, + ) + ) + except RealFeatureLedgerPromotionError as exc: + errors.append(str(exc)) + return { + "schema": VALIDATION_SCHEMA, + "status": "valid" if not errors else "invalid", + "local_real_feature_ledger_promotion_verified": not errors, + "promotion_persisted_to_gda_control": ( + not errors + and evidence is not None + and evidence.get("promotion_persisted_to_gda_control") is True + ), + "platform_run_succeeded": False, + "production_ready": False, + "contract_sha256": contract["contract_sha256"], + "evidence_sha256": evidence.get("evidence_sha256") if evidence else None, + "errors": errors, + } + + +def _write_json(path: Path, payload: Mapping[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, ensure_ascii=True, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + contract = subparsers.add_parser("contract") + contract.add_argument( + "--source-evidence", type=Path, default=DEFAULT_SOURCE_EVIDENCE_PATH + ) + validate = subparsers.add_parser("validate") + validate.add_argument( + "--source-evidence", type=Path, default=DEFAULT_SOURCE_EVIDENCE_PATH + ) + validate.add_argument("--evidence", type=Path, default=DEFAULT_EVIDENCE_PATH) + rehearse = subparsers.add_parser("rehearse") + rehearse.add_argument("--database-url", required=True) + rehearse.add_argument( + "--source-evidence", type=Path, default=DEFAULT_SOURCE_EVIDENCE_PATH + ) + rehearse.add_argument("--output", type=Path, default=DEFAULT_EVIDENCE_PATH) + args = parser.parse_args(argv) + try: + if args.command == "contract": + report = build_contract_report( + source_evidence_path=args.source_evidence + ) + elif args.command == "rehearse": + report = run_postgres_rehearsal( + args.database_url, + source_evidence_path=args.source_evidence, + ) + _write_json(args.output, report) + else: + report = build_validation_report( + source_evidence_path=args.source_evidence, + evidence_path=args.evidence, + ) + print(json.dumps(report, ensure_ascii=True, indent=2, sort_keys=True)) + return 0 if not report.get("errors") else 1 + except ( + OSError, + TypeError, + ValueError, + RealFeatureLedgerPromotionError, + GatewayConflictError, + GatewayValidationError, + ) as exc: + print( + json.dumps( + {"status": "blocked", "error": type(exc).__name__}, + ensure_ascii=True, + sort_keys=True, + ) + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/platform_truth.py b/data_agent/platform_truth.py index 774817d4..70d19f5b 100644 --- a/data_agent/platform_truth.py +++ b/data_agent/platform_truth.py @@ -1018,6 +1018,26 @@ def _config( ), "Versioned GDA Control promotion, protected identity/storage and production ingestion", ), + RuntimeSpec( + "metadata_real_feature_ledger_promotion_rehearsal", + "real_feature_ledger_promotion_rehearsal", + "governed", + "evidence_durable", + "temporary PostgreSQL control ledger + committed local evidence", + "metadata-platform", + "local_verification_only", + ( + "data_agent/metadata_fabric_real_feature_ledger_promotion.py", + "scripts/metadata-fabric-real-feature-ledger-promotion.sh", + ), + ( + ( + "data_agent/metadata_fabric_real_feature_ledger_promotion.py", + "def run_postgres_rehearsal", + ), + ), + "Retained output material, terminal success evidence and production promotion", + ), RuntimeSpec( "datalake_monitor", "monitor_loop", diff --git a/data_agent/test_metadata_fabric_real_feature_ledger_promotion.py b/data_agent/test_metadata_fabric_real_feature_ledger_promotion.py new file mode 100644 index 00000000..de0587e2 --- /dev/null +++ b/data_agent/test_metadata_fabric_real_feature_ledger_promotion.py @@ -0,0 +1,123 @@ +import json +from copy import deepcopy + +import pytest +from pydantic import ValidationError + +from data_agent import metadata_fabric_real_feature_ingestion as m322 +from data_agent import metadata_fabric_real_feature_ledger_promotion as promotion +from data_agent.platform_contracts import quality_result_fingerprint + + +def _source() -> dict: + return json.loads( + promotion.DEFAULT_SOURCE_EVIDENCE_PATH.read_text(encoding="utf-8") + ) + + +def test_checked_m3_22_candidates_build_one_content_bound_promotion(): + source = _source() + bundle = promotion.build_promotion(source) + + assert bundle.authority_resource.resource_kind == "data_product" + assert bundle.authority_resource.authority_system == "gravitino" + assert bundle.output_resource_version.resource_version_id == ( + promotion.OUTPUT_RESOURCE_VERSION_ID + ) + assert bundle.output_resource_version.content_sha256 == ( + source["observation"]["plan"]["output_content_sha256"] + ) + assert bundle.output_artifact.run_id == promotion.RUN_ID + assert bundle.quality_result.evaluated_by == m322.QUALITY_EVALUATOR + assert bundle.quality_result.evaluated_by != ( + bundle.output_resource_version.created_by + ) + assert bundle.lineage_event.source_resource_version_id == ( + promotion.SOURCE_RESOURCE_VERSION_ID + ) + assert bundle.lineage_event.target_resource_version_id == ( + promotion.OUTPUT_RESOURCE_VERSION_ID + ) + + +def test_promotion_prerequisites_correlate_without_fabricating_authorization(): + source = _source() + bundle = promotion.build_promotion(source) + prerequisites = promotion.build_prerequisites(source, bundle) + + assert prerequisites.run.status.value == "accepted" + assert prerequisites.run.state_version == 0 + assert prerequisites.run.policy_refs is None + assert prerequisites.run.config_fingerprint == ( + source["observation"]["authorization"]["authorization_sha256"] + ) + assert prerequisites.run.input_bindings[0].resource_version_id == ( + promotion.SOURCE_RESOURCE_VERSION_ID + ) + assert prerequisites.definition_registration.definition.output_contract[ + "platform_run_terminal_success" + ] is False + assert prerequisites.output_resource == bundle.authority_resource + + +def test_m3_22_evidence_tampering_is_rejected_before_promotion(): + source = deepcopy(_source()) + source["observation"]["output_contracts"]["output_resource_version"][ + "content_sha256" + ] = "0" * 64 + + with pytest.raises( + promotion.RealFeatureLedgerPromotionError, + match="M3-22 evidence is invalid", + ): + promotion.build_promotion(source) + + +def test_promotion_contract_rejects_quality_evaluator_impersonation(): + bundle = promotion.build_promotion(_source()) + values = bundle.model_dump(mode="python") + quality_values = bundle.quality_result.model_dump(mode="python") + quality_values["evaluated_by"] = bundle.output_resource_version.created_by + quality_values["result_sha256"] = quality_result_fingerprint( + **{ + key: value + for key, value in quality_values.items() + if key not in {"quality_result_id", "result_sha256"} + } + ) + values["quality_result"] = bundle.quality_result.__class__(**quality_values) + + with pytest.raises(ValidationError, match="quality evaluation must be independent"): + promotion.RunOutputLedgerPromotion.model_validate(values) + + +def test_promotion_contract_rejects_lineage_manifest_drift(): + bundle = promotion.build_promotion(_source()) + values = bundle.model_dump(mode="python") + lineage = bundle.lineage_event.model_copy( + update={"facets": {**bundle.lineage_event.facets, "feature_count": 21}} + ) + values["lineage_event"] = lineage + + with pytest.raises(ValidationError, match="lineage facets do not match"): + promotion.RunOutputLedgerPromotion.model_validate(values) + + +def test_static_contract_is_source_bound_and_non_terminal(): + report = promotion.build_contract_report() + + assert report["status"] == "valid" + assert report["errors"] == [] + assert report["source_evidence_sha256"] == promotion.SOURCE_EVIDENCE_SHA256 + assert report["source_contract_sha256"] == promotion.SOURCE_CONTRACT_SHA256 + assert report["atomic_write_order"] == [ + "resource_version", + "output_artifact", + "quality_evidence_artifact", + "quality_result", + "lineage_event", + ] + assert report["requires_preexisting_output_authority"] is True + assert report["partial_preexisting_state_rejected"] is True + assert report["platform_run_terminal_success"] is False + assert report["writes_to_legacy"] is False diff --git a/data_agent/test_metadata_fabric_real_feature_ledger_promotion_postgres.py b/data_agent/test_metadata_fabric_real_feature_ledger_promotion_postgres.py new file mode 100644 index 00000000..50f3a50e --- /dev/null +++ b/data_agent/test_metadata_fabric_real_feature_ledger_promotion_postgres.py @@ -0,0 +1,115 @@ +import json +import os +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.engine import make_url + +from data_agent import metadata_fabric_real_feature_ledger_promotion as promotion +from data_agent.platform_gateway import GatewayConflictError, PlatformGateway + +DATABASE_URL = os.environ.get("DATABASE_URL") + + +def _temporary_database_url(prefix: str) -> tuple[object, str, str]: + admin_url = make_url(DATABASE_URL) + admin_engine = create_engine(admin_url, isolation_level="AUTOCOMMIT") + with admin_engine.connect() as connection: + if not connection.exec_driver_sql( + "SELECT rolsuper FROM pg_roles WHERE rolname = current_user" + ).scalar_one(): + admin_engine.dispose() + pytest.skip("M3-23 PostgreSQL test requires a superuser") + database_name = f"{prefix}_{uuid4().hex}" + connection.exec_driver_sql(f'CREATE DATABASE "{database_name}"') + database_url = admin_url.set(database=database_name).render_as_string( + hide_password=False + ) + return admin_engine, database_name, database_url + + +def _drop_temporary_database(admin_engine, database_name: str) -> None: + with admin_engine.connect() as connection: + connection.execute( + text( + """ + SELECT pg_terminate_backend(pid) + FROM pg_stat_activity + WHERE datname = :database_name + AND pid <> pg_backend_pid() + """ + ), + {"database_name": database_name}, + ) + connection.exec_driver_sql(f'DROP DATABASE "{database_name}"') + admin_engine.dispose() + + +@pytest.mark.skipif(not DATABASE_URL, reason="DATABASE_URL is not configured") +def test_real_m3_22_candidates_promote_atomically_on_fresh_postgres(): + admin_engine, database_name, database_url = _temporary_database_url( + "gda_real_feature_promotion" + ) + try: + evidence = promotion.run_postgres_rehearsal(database_url) + + assert promotion.validate_rehearsal_evidence(evidence) == [] + assert evidence["first_promotion_created"] is True + assert evidence["replay_promotion_created"] is False + assert evidence["failure_injection_rollback_verified"] is True + assert evidence["candidate_row_counts"] == { + "resource_versions": 1, + "artifacts": 2, + "quality_results": 1, + "lineage_events": 1, + } + assert evidence["platform_run_status"] == "accepted" + assert evidence["success_finalization_rejected"] is True + assert evidence["promotion_persisted_to_gda_control"] is True + assert evidence["platform_run_succeeded"] is False + finally: + _drop_temporary_database(admin_engine, database_name) + + +@pytest.mark.skipif(not DATABASE_URL, reason="DATABASE_URL is not configured") +def test_partial_preexisting_promotion_is_rejected_without_new_rows(): + admin_engine, database_name, database_url = _temporary_database_url( + "gda_partial_real_feature_promotion" + ) + engine = None + try: + source = json.loads( + promotion.DEFAULT_SOURCE_EVIDENCE_PATH.read_text(encoding="utf-8") + ) + bundle = promotion.build_promotion(source) + prerequisites = promotion.build_prerequisites(source, bundle) + engine = create_engine(database_url) + promotion._apply_migrations(engine) + gateway = PlatformGateway(engine) + promotion._register_prerequisites( + gateway, + prerequisites, + include_output_authority=True, + ) + gateway.register_resource_version(bundle.output_resource_version) + + with pytest.raises( + GatewayConflictError, + match="partial pre-existing state", + ): + promotion.RunOutputLedgerPromoter(gateway).promote(bundle) + + assert promotion._candidate_counts(engine, bundle) == { + "resource_versions": 1, + "artifacts": 0, + "quality_results": 0, + "lineage_events": 0, + } + assert gateway.get_run(promotion.TENANT, promotion.RUN_ID).status.value == ( + "accepted" + ) + finally: + if engine is not None: + engine.dispose() + _drop_temporary_database(admin_engine, database_name) diff --git a/data_agent/test_platform_truth.py b/data_agent/test_platform_truth.py index f31b5445..5f90b932 100644 --- a/data_agent/test_platform_truth.py +++ b/data_agent/test_platform_truth.py @@ -297,6 +297,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_real_feature_ledger_promotion_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-069-atomic-real-feature-output-ledger-promotion.md b/docs/architecture-decisions/adr-069-atomic-real-feature-output-ledger-promotion.md new file mode 100644 index 00000000..724dfa76 --- /dev/null +++ b/docs/architecture-decisions/adr-069-atomic-real-feature-output-ledger-promotion.md @@ -0,0 +1,90 @@ +# ADR-069: Atomic real-feature output ledger promotion + +**Status**: Accepted + +**Date**: 2026-07-31 + +**Decision owners**: Data Platform, Metadata Platform, Data Governance, GIS Platform, Security, Platform Architecture + +**Related decisions**: [ADR-047](adr-047-deterministic-metadata-fabric-ingestion-projection.md) · [ADR-068](adr-068-local-authorized-real-feature-iceberg-ingestion.md) · [ADR-022](adr-022-platform-control-gateway.md) + +## Context + +M3-22 produced a path-free output ResourceVersion, output Artifact, Iceberg metadata evidence Artifact, independently evaluated passed QualityResult and source-to-output LineageEvent for the real 20-feature Chongqing cultural-district slice. It deliberately kept every object outside GDA Control and left the PlatformRun non-terminal. + +Calling the existing public `register_resource_version`, `record_artifact`, `record_quality_result` and `record_lineage` methods sequentially would commit one transaction per call. A process failure could therefore leave an authority-looking partial ledger. The promotion must also reject a missing or drifted output Resource authority record, exact replay must be a no-op, and the existing PlatformRun success function must remain the only terminal success authority. + +The M3-22 runtime and object-store material were deleted after its local rehearsal. M3-23 can prove the control-ledger mechanism against the checked evidence, but it cannot turn the deleted local S3 location into retained production material. + +## Options considered + +| Option | Benefit | Cost or risk | Decision | +|---|---|---|---| +| Call existing public gateway writes in sequence | No gateway refactor | Partial commits are observable after failure | Rejected | +| Add a new `SECURITY DEFINER` database function | Strong database-owned entry point | Duplicates existing contract validation and adds migration/privilege surface without a current need | Deferred | +| Compose a dedicated promoter over one gateway transaction | Reuses current RLS, foreign keys, append-only triggers and grants without changing the evidence-bound gateway module | All callers must use the dedicated promoter for this bundle | Chosen | + +## Decision + +### 1. Promote one content-bound bundle + +`RunOutputLedgerPromotion` binds exactly one pre-existing authority Resource to: + +- one output ResourceVersion; +- one output Artifact with the same content SHA and Run; +- one distinct quality evidence Artifact; +- one passed QualityResult whose evaluator differs from the output creator and whose rule/metrics match the evidence Artifact; +- one LineageEvent from a Run input ResourceVersion to the output version and output Artifact. + +Tenant, Run, DefinitionVersion, ResourceVersion, Artifact and content identities must match before any write. + +### 2. Use one existing PostgreSQL transaction boundary + +`RunOutputLedgerPromoter.promote` is isolated in the M3-23 module so adding this bundle does not invalidate the earlier gateway-bound evidence chain. It composes the existing `PlatformGateway` transaction, load and ResourceVersion-write primitives, validates the pre-existing Resource authority, source ResourceVersion, DefinitionVersion and accepted/reconciling PlatformRun, then writes in foreign-key order: + +1. ResourceVersion; +2. output Artifact; +3. quality evidence Artifact; +4. QualityResult; +5. LineageEvent. + +The promoter reuses the existing `_transaction` boundary with `SET LOCAL ROLE gda_control_gateway` and tenant context. Artifact, QualityResult and LineageEvent inserts remain private to that transaction coordinator; no public single-record gateway method is called mid-transaction. No legacy table is written and no new database privilege is granted. + +All five writes must report the same creation state. `true` for all is the first commit; `false` for all is an exact replay. Mixed creation states mean partial pre-existing state and fail the entire new transaction. Existing conflicting identity/content also fails closed. + +### 3. Keep authority and terminal success separate + +The output Resource authority record is a prerequisite rather than an implicit side effect of candidate promotion. M3-23 records its local Gravitino/Iceberg identity and explicitly marks the material as not retained and not production-ready. + +M3-22 did not retain complete PolicyDecision and Approval Artifact payloads. M3-23 therefore creates an accepted Run correlation with the checked M3-22 authorization fingerprint as `config_fingerprint`; it does not fabricate those missing Artifacts or claim that M3-22 authorization was persisted to GDA Control. + +The promotion transaction never calls `finalize_platform_run_success`. The existing terminal gate still lacks a successful FrameworkAttemptObservation. In addition, the M3-22 Iceberg metadata evidence Artifact was created by the ingestion workload, not by the independent quality evaluator required by the success function. An explicit finalization attempt must fail and leave the Run at `accepted`, state version zero, with only its initial event. + +### 4. Treat the result as local control-plane evidence + +The real PostgreSQL rehearsal uses a fresh temporary database, proves rollback through an injected failure before QualityResult append, proves exact first/replay behavior, FORCE RLS, cross-tenant isolation, minimum grants and direct UPDATE/DELETE rejection, then deletes the database. + +`promotion_persisted_to_gda_control=true` means the checked bundle committed to that real temporary GDA Control database. It does not mean the local Iceberg material, database or provider runtime was retained, deployed to staging or promoted to production. + +## Verification + +The M3-23 rehearsal recorded: + +- source M3-22 evidence SHA `42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899`; +- output content SHA `bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618`; +- promotion SHA `404b6e4e5d8194f092bd83ef99cbf2d1d727015b926cd438a79eb0210f969a22`; +- injected failure rollback counts of zero for all five candidate categories; +- first promotion `created=true`, exact replay `created=false`, with row counts `1 ResourceVersion + 2 Artifacts + 1 QualityResult + 1 LineageEvent`; +- missing authority, cross-tenant read/direct insert and eight direct mutation attempts rejected; +- PlatformRun `accepted@0`, one initial event and explicit success finalization rejection; +- contract SHA `bd21c81925f66acdfecca5cabd78651f31deab4165da2ccd6900c4e5796e5735` and evidence SHA `f6efea5000791dec1716a8354a8e39a8425b083ca4d409f4bcb61f0e7e03580d`. + +## Consequences + +**Positive**: a real GIS output can now cross from checked provider evidence into the version, Artifact, quality and lineage ledgers as one idempotent transaction without weakening RLS, append-only or terminal success authority. + +**Negative**: M3-23 proves a local control-plane commit against historical evidence. Its target material and database are not retained. The correlated Run does not contain persisted M3-22 policy/approval Artifacts and cannot be finalized successfully. + +**Mitigation**: the next retained staging slice must produce policy/approval, provider success observation, independently created quality evidence and output material in durable selected storage before calling the same promotion method and the existing success gate. + +**Revisit trigger**: add a database-owned promotion function only if another trusted writer must perform the same bundle commit without the Python gateway, or if production concurrency shows that gateway transaction serialization is insufficient. diff --git a/docs/evidence/metadata-fabric-real-feature-ledger-promotion-2026-07-31.json b/docs/evidence/metadata-fabric-real-feature-ledger-promotion-2026-07-31.json new file mode 100644 index 00000000..954d51e9 --- /dev/null +++ b/docs/evidence/metadata-fabric-real-feature-ledger-promotion-2026-07-31.json @@ -0,0 +1,67 @@ +{ + "authority_locator": "gda_chongqing_m3_22/lakehouse/cultural_heritage/cultural_districts", + "authority_system": "gravitino", + "candidate_row_counts": { + "artifacts": 2, + "lineage_events": 1, + "quality_results": 1, + "resource_versions": 1 + }, + "contract_sha256": "bd21c81925f66acdfecca5cabd78651f31deab4165da2ccd6900c4e5796e5735", + "cross_tenant_direct_insert_blocked": true, + "cross_tenant_read_blocked": true, + "definition_version_id": "a9000000-0000-4000-8000-000000000008", + "direct_mutations_blocked": { + "artifact_delete": true, + "artifact_update": true, + "lineage_event_delete": true, + "lineage_event_update": true, + "quality_result_delete": true, + "quality_result_update": true, + "resource_version_delete": true, + "resource_version_update": true + }, + "durable_catalog_verified": false, + "errors": [], + "evidence_sha256": "f6efea5000791dec1716a8354a8e39a8425b083ca4d409f4bcb61f0e7e03580d", + "exact_replay_verified": true, + "failure_injection_rollback_verified": true, + "first_promotion_created": true, + "force_rls_verified": true, + "m322_authorization_persisted_to_gda_control": false, + "minimum_grants_verified": true, + "missing_authority_rejected": true, + "output_content_sha256": "bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618", + "output_material_retained": false, + "output_resource_urn": "gda://metadata-authorization-local/data_product/chongqing-cultural-districts-iceberg", + "output_resource_version_id": "a6000000-0000-4000-8000-000000000002", + "platform_run_event_count": 1, + "platform_run_state_version": 0, + "platform_run_status": "accepted", + "platform_run_succeeded": false, + "production_ingestion_verified": false, + "production_object_store_verified": false, + "production_ready": false, + "promotion_persisted_to_gda_control": true, + "promotion_sha256": "404b6e4e5d8194f092bd83ef99cbf2d1d727015b926cd438a79eb0210f969a22", + "protected_workload_identity_verified": false, + "replay_promotion_created": false, + "rollback_candidate_counts": { + "artifacts": 0, + "lineage_events": 0, + "quality_results": 0, + "resource_versions": 0 + }, + "run_id": "a9000000-0000-4000-8000-000000000009", + "schema": "gda.real_feature_ledger_promotion_evidence.v1", + "source_absolute_path_committed": false, + "source_dataset_committed": false, + "source_evidence_schema": "gda.real_feature_ingestion_evidence.v1", + "source_evidence_sha256": "42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899", + "source_feature_payload_committed": false, + "source_resource_version_id": "a6000000-0000-4000-8000-000000000001", + "status": "local_real_feature_ledger_promotion_verified", + "success_finalization_rejected": true, + "tenant_id": "metadata-authorization-local", + "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 aacbfce2..4b81f03e 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-22（本地真实要素 JDBC/S3 ingestion 已验证，权威晋级与生产验证待执行） +### 4.8 Metadata Fabric Bridge M1 + M2 + M3-23（本地真实要素控制账本原子晋级已验证，持久 material 与生产验证待执行） 第八块回到 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： @@ -242,10 +242,11 @@ Temporal 继续保持目标组件状态，不在这一包并行接入。OpenMeta 37. [ADR-066](architecture-decisions/adr-066-runtime-bound-durable-active-metadata-promotion.md) 保持 M3-19 binding schema/ledger/evidence 不变，将同一重庆 ResourceVersion 投影到隔离 JDBC metadata + warehouse PVC 的 Gravitino target，并把 logical provider ref 与 cluster/namespace/Service/StatefulSet/PVC/image identity 组合为独立 promotion candidate。受限 Basic principal 首次只执行 1 个 `gravitino.table.create`；即时 replay 与 PostgreSQL -> Gravitino 有序 restart 后的第一次 replay 都为 `no_op/0 mutations`，两次 Pod UID 变化而稳定 runtime/PVC identity 与 table projection 不变。logical binding SHA 为 `8c312db37bfe92e034bcdcb7a3c35847c81e862c74a3437970def1007af42750`，runtime binding SHA 为 `a78975311fc34abd76fa41dea581594806b3d18ed364ba518cfc44c4204822f7`，promotion candidate SHA 为 `bb6672cb7f98fa53305e17bbca2cb5b3756d4a335a94d79114fb4184273871d1`，contract/evidence SHA 分别为 `307f2d4390028589c0f38be859c53826bd149d7f2a133b14488230d4f5ff6eb8` / `53773e9417668e03ad3ab2b5c3cdbd627fb3bc397d63c5860755ec5318eebe8b`。candidate 未写 GDA Control，namespace/PVC 已清理；`durable_catalog_verified=false`、`production_object_store_verified=false`、`production_ready=false`。 38. [ADR-067](architecture-decisions/adr-067-object-store-runtime-bound-active-metadata-promotion.md) 以 M3-20 promotion candidate 为不可变 predecessor，在 M3-10 跨节点 MinIO runtime 中创建独立 `gda_chongqing_m3_21` JDBC/S3 target。runtime binding 同时包含 Gravitino/MinIO Service、PostgreSQL/Gravitino/MinIO StatefulSet、PostgreSQL/MinIO PVC、镜像、节点分离与 S3 warehouse/endpoint/bucket，且 Gravitino 无 warehouse PVC。受限 principal 首次仅 1 个 table create，即时及 PostgreSQL -> Gravitino restart 后首个 replay 均为 `no_op/0`；S3 直读在精确 prefix 下只见 1 个 Iceberg metadata JSON，无 data/manifest，key/ETag/body SHA/表 schema 重启前后不变。predecessor/logical/runtime/promotion SHA 分别为 `bb6672cb7f98fa53305e17bbca2cb5b3756d4a335a94d79114fb4184273871d1`、`614ce5e4c45dba1437dc888cbd79b2d58954184113a62c20170ab84b5570d9e1`、`dd63917b6354a2e92853763ddc3e3a981cb40717f84c0f819b1a4e6844ae100b`、`63812c311b3f239bc6a944748c4ff384250eb9c9ed9009d3384fc699f1d3eaa9`；contract/evidence SHA 为 `b1a2db34a70eaa7dd55da1d6c85da9f420c755c71868aafe7972e3794034a6cc` / `d73754c53cf16d888aa345baa5d079cc7fd98d8b84db747f52188c1a69bf1628`。candidate 未落账、未 ingest feature rows，namespace/PVC/port-forward 已清理；生产对象存储与 readiness 仍为 `false`。 39. [ADR-068](architecture-decisions/adr-068-local-authorized-real-feature-iceberg-ingestion.md) 复用同一重庆 20-feature EPSG:4490 bundle 和 M3-21 predecessor，在独立 `gda_chongqing_m3_22` target 中由受限 Gravitino principal 创建八列表，再由一个内容与授权指纹绑定的 Spark `3.5.0` + Sedona `1.9.0` Job 写入。六项空间质量计数均为 20；首次执行 `appended/1`，产生 1 个 snapshot 和 1 个 20-row Parquet，即时 replay 为 `no_op/0` 且 row/snapshot/file readback 不变。S3 直读为 1 data + 2 metadata + 2 manifest，并构造独立 output ResourceVersion、Artifact、passed QualityResult 和 LineageEvent candidates。row-set/output/contract/evidence SHA 分别为 `c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df`、`bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618`、`af211f2d2f4830decb9ffe369cd9e7ec2c9349c2e2c8bd789347a6fdc288e1dc`、`42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899`。源路径、BSM、WKB 和 credential 不入 evidence，namespace/PVC/port-forward 已清理；candidates 未落 GDA Control，Run 未终局，生产 ingestion/readiness 仍为 `false`。 +40. [ADR-069](architecture-decisions/adr-069-atomic-real-feature-output-ledger-promotion.md) 将 M3-22 的 output ResourceVersion、output Artifact、quality evidence Artifact、独立 passed QualityResult 与 source-to-output LineageEvent 组合为 `RunOutputLedgerPromotion`，要求 output authority Resource、源版本、Definition 和 accepted/reconciling Run 已存在，再由独立 promoter 按外键顺序在一个 PlatformGateway PostgreSQL 事务中追加，且不改变此前证据绑定的 gateway 模块。缺 authority、混合半状态、跨租户读/直写和八个 UPDATE/DELETE 均拒绝；QualityResult 前故障注入后五类候选计数全为 0，首次 `created=true`、精确 replay `created=false`，最终只形成 `1 version + 2 artifacts + 1 quality + 1 lineage`。Run 保持 `accepted@0`，显式 success finalization 被既有 gate 拒绝。promotion/contract/evidence SHA 分别为 `404b6e4e5d8194f092bd83ef99cbf2d1d727015b926cd438a79eb0210f969a22`、`bd21c81925f66acdfecca5cabd78651f31deab4165da2ccd6900c4e5796e5735`、`f6efea5000791dec1716a8354a8e39a8425b083ca4d409f4bcb61f0e7e03580d`。临时 GDA Control 数据库和 M3-22 material 均未保留，M3-22 authorization Artifacts 未伪造补写；生产 ingestion/readiness 仍为 `false`。 -M3-22 已证明本地同主机双节点上一个受授权真实 feature slice 的空间质量、Iceberg 单次 append、即时 no-op replay、S3 直读和 path-free output candidates；它不替代权威 GDA Control promotion、Run 成功终局、staging/大规模 ingestion、生产对象存储 attestation、独立 failure domain、KMS/TLS/workload identity、备份/PITR 或 tenant isolation。 +M3-23 已证明上述真实 feature slice 的 path-free candidates 可在真实 PostgreSQL 中以单事务、精确重放和 fail-closed 半状态规则晋级到临时 GDA Control；它不替代可保留的 output material、完整授权 Artifact、独立质量 evidence provenance、成功 observation、Run 成功终局、staging/大规模 ingestion、生产对象存储 attestation、独立 failure domain、KMS/TLS/workload identity、备份/PITR 或 tenant isolation。 -此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file；M3-13 只证明单次本地 append 在 provider 200 响应丢失并映射为 commit-state-unknown 后，可以由即时 table readback 判定 committed 且不重提，不覆盖持久 controller、进程崩溃、并发写或任意 mutation；M3-14 只证明 ResourceVersion 注册与 Active Metadata 事件在本地 PostgreSQL 同事务创建，并验证租户/workload scoped claim/retry/complete；M3-15 只证明默认零副本 managed consumer 的代码/部署边界，以及本地 PostgreSQL 中 inert activation request 与 event completion 的原子性；M3-16 只证明本地真实数据 content fingerprint、证据绑定授权与 pending command 的 PostgreSQL 原子性；M3-17 只证明本地 standalone 中既有 consumer/adapter 的真实 submission、精确 correlation read-back 和 provider success observation；M3-18 只证明同一 Docker Desktop 主机上 scheduler 通过 ephemeral HTTP executor 触发 bootstrap-admin/unauthenticated providers 的一次创建和同进程零写 replay；M3-19 只证明同一主机上 exact OpenMetadata + absent Gravitino 的受限修复、即时 no-op replay 和临时 PostgreSQL binding commit，且 `memory` catalog reset 明确不等于 durable catalog recovery；M3-22 只证明短生命周期本地 runtime 中一份 20-row 真实 slice 的授权写入、质量 readback 和 path-free candidates，未向 GDA Control 晋级或形成 Run 成功终局。生产持久 binding deployment、ResourceVersion 和 legacy authority 都未切换；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 +此处 M1 只证明静态合同和只读 HTTP 边界；M2a 只证明本地 live foundation 与 PVC 重挂载连续性；M2b-1/M2b-2 分别限定在同集群新 PVC 和同集群隔离 repository；M2b-3 的 `local_cross_cluster_recovery_verified=true` 只限定在 `local_same_host_distinct_kubernetes_clusters_external_s3_repository`；M2c-1/M2c-2/M2c-3 分别限定本地 provider metrics、临时双周期 OTel 和单 job scrape recovery；M2c-4/M2d-2 只证明 production observability/NetworkPolicy profile 与 attestation 合同可校验；M2d-1 只证明本地两节点 kindnet 的隔离合成流量；M3-1 的 terminal evidence 与 M3-2 的 PolicyDecision/Approval 仍是 deterministic local fixtures。M3-2 只把 projection 写入本地 provider 并证明 retained target 的单次零写入 replay；M3-3 只把该本地 evidence 对应的 binding 写入临时 GDA Control 账本；M3-4 只向无认证 loopback receiver 发送精确 candidate 并验证 503 后幂等恢复；M3-5 只证明 OpenMetadata 在 provider 强制默认 role 之上的项目新增 grant 限定为 `table/Create`，以及本地 JWT 轮换/吊销和越权拒绝；M3-6 只证明隔离 Gravitino Basic IdP 的 bounded table-create、catalog-create 拒绝、登录轮换/吊销和完整清理；M3-7 只证明 pending production identity profile、profile-bound attestation 和派生 claim 的 fail-closed 合同可校验，没有部署或证明真实身份路径；M3-8 只证明同一 Docker Desktop 集群内 Basic 用户、JDBC metadata 与 file warehouse PVC 在受控 Pod restart 后连续；M3-9 只证明同节点共享 RWO PVC 的 Spark interoperability；M3-10 移除了该共享 PVC，并证明同一 Docker Desktop 主机/集群内 Spark 与 MinIO 的跨节点 S3-compatible 互操作，但不证明生产云对象存储、独立 failure domain、持久 identity binding、Flink 或完整 engine conformance；M3-11 只冻结 provider-neutral production object-store profile、精确 attestation binding 与 fail-closed claims，没有选择 provider、部署 bucket/KMS/policy 或提交真实 attestation；M3-12 只证明同一本地路径的 pre-forward commit failure 不改变可见 table state，随后一次显式重试产生一个新 snapshot/row，且无孤儿 data file；M3-13 只证明单次本地 append 在 provider 200 响应丢失并映射为 commit-state-unknown 后，可以由即时 table readback 判定 committed 且不重提，不覆盖持久 controller、进程崩溃、并发写或任意 mutation；M3-14 只证明 ResourceVersion 注册与 Active Metadata 事件在本地 PostgreSQL 同事务创建，并验证租户/workload scoped claim/retry/complete；M3-15 只证明默认零副本 managed consumer 的代码/部署边界，以及本地 PostgreSQL 中 inert activation request 与 event completion 的原子性；M3-16 只证明本地真实数据 content fingerprint、证据绑定授权与 pending command 的 PostgreSQL 原子性；M3-17 只证明本地 standalone 中既有 consumer/adapter 的真实 submission、精确 correlation read-back 和 provider success observation；M3-18 只证明同一 Docker Desktop 主机上 scheduler 通过 ephemeral HTTP executor 触发 bootstrap-admin/unauthenticated providers 的一次创建和同进程零写 replay；M3-19 只证明同一主机上 exact OpenMetadata + absent Gravitino 的受限修复、即时 no-op replay 和临时 PostgreSQL binding commit，且 `memory` catalog reset 明确不等于 durable catalog recovery；M3-22 只证明短生命周期本地 runtime 中一份 20-row 真实 slice 的授权写入、质量 readback 和 path-free candidates；M3-23 只证明这些 candidates 在临时 PostgreSQL 的原子晋级与 replay/rollback/security 边界，没有保留 output material、数据库、完整授权 Artifact 或成功 observation，也没有形成 Run 成功终局。生产持久 binding deployment、ResourceVersion 和 legacy authority 都未切换；生产对象存储、双 provider/生产最小权限、protected workload identity、OIDC、TLS、生产持久 catalog、tenant isolation、真实 receiver/alert/SLO、受保护 provider policy、生产故障注入、source-loss recovery、cancel/reconcile/lineage、完整 Spark/Flink conformance、生产 ingest、四项 production gate 和 `production_ready` 仍为 `false`。 ## 5. 重新评估条件 diff --git a/docs/roadmap.md b/docs/roadmap.md index 76bea67f..ac12c50c 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -419,6 +419,8 @@ AR-0 Architecture / Schema / Runtime Truth 当前 Metadata Fabric 证据边界：M1 只读 bridge 合同已验证；ADR-037 至 ADR-046 分别覆盖本地 foundation/recovery/metrics/network-policy 演练与 production readiness contracts；ADR-047 至 ADR-050 已依次建立 deterministic projection plan、本地双 provider replay、tenant-scoped binding ledger 与本地 OpenLineage 幂等 wire delivery；ADR-051 以临时非管理员 OpenMetadata bot 证明项目专用 grant 只有 `table/Create`、`policy/Create` 被 403 拒绝，且 JWT 轮换/吊销后旧值/当前值均返回 401；ADR-052 又在隔离 Gravitino `1.3.0` Basic IdP 中证明 bounded user 的 `USE_CATALOG`、`USE_SCHEMA`、`CREATE_TABLE` 范围、catalog-create 403、密码轮换和用户吊销；ADR-053 将生产 OIDC federation、双 provider integration/workload identity、最小权限、TLS/mTLS、持久 Gravitino catalog、tenant isolation、运营责任和新鲜 protected attestation 冻结为 fail-closed readiness contract。Gravitino `1.3.0` 镜像只发现 Basic IdP，不假设 native OIDC；当前 profile 仍有 40 个外部 blockers 且未提交真实 attestation。`local_openmetadata_minimum_privilege_verified=true` 与 `local_gravitino_minimum_privilege_verified=true` 都只描述各自临时 provider rehearsal；M3-2 ingestion 仍使用 bootstrap admin，Gravitino probe catalog 仍是 memory catalog。因此 `provider_minimum_privilege_verified`、protected workload identity、OIDC、TLS、持久 catalog、生产 ingestion/conformance、生产 lineage receiver、`production_identity_gate_passed` 与 `production_ready` 仍为 false。 +M3-22/M3-23 已把一份真实重庆 20-feature EPSG:4490 slice 从受授权 Spark/Sedona + JDBC/S3 Iceberg ingestion 推进到临时 GDA Control 的原子 `ResourceVersion + 2 Artifacts + QualityResult + LineageEvent` 晋级，并验证失败整笔回滚、精确 replay、FORCE RLS、跨租户和 direct mutation 拒绝。该结果仍限定在已删除的本地 material 与临时 PostgreSQL；Run 保持 `accepted@0`，完整 PolicyDecision/Approval Artifact、成功 observation、独立 quality evidence provenance、持久 staging material 和生产 identity/storage/tenant attestation 仍是下一门槛。 + ### AR-2 — Source, Ingestion and Geospatial Lakehouse Vertical Slice（P0） **依赖**：AR-1 两个控制面的最小合同通过故障注入和隔离验收。 @@ -726,7 +728,7 @@ AR-4 parity/control gate 退出前暂停以下主线扩张： | 阶段 | 状态 | 下一证据 | |---|---|---| | AR-0 Architecture/Schema/Runtime Truth Freeze | `in_progress` | 全环境 schema/config fingerprint、迁移 fail-closed、事实清单、storage/compute/GIS serving provider profile/capability、ADR-017 benchmark、owner/SLO 和首条数据/服务验收集冻结 | -| AR-1 Unified Metadata + Orchestration Control Planes | `in_progress` | controlled gateway、DolphinScheduler adapter、Metadata Fabric M1/M2a/M2b、M2c-1 provider metrics、M2c-2 本地临时 OTel pipeline 与 M2c-3 本地 scrape failure/recovery 已验证；下一证据是 source host/cluster 外的生产 recovery、持久 metrics backend/TLS/tenant/真实 alert delivery/SLO、OIDC、NetworkPolicy enforcement、升级回滚/registry provenance，以及受控 ingestion/replay 与无双写验收 | +| AR-1 Unified Metadata + Orchestration Control Planes | `in_progress` | controlled gateway、DolphinScheduler adapter、Metadata Fabric 本地 recovery/metrics/policy/identity/interoperability、真实 feature ingestion 与临时 GDA Control 原子 output promotion 已验证；下一证据是可保留 staging material、完整授权/成功/独立质量 provenance、source host/cluster 外 recovery、持久 metrics/TLS/tenant/alert/SLO、OIDC、NetworkPolicy、升级回滚/registry provenance 与无双写验收 | | AR-2 Source/Ingestion + Geospatial Lakehouse Vertical Slice | `planned` | 三类代表源、`DriveTransfer` 云盘客户端和大文件恢复通过统一控制面；默认湖仓、轻量存算一体及 Azure 代表 adapter 通过 provider conformance 与 Raw -> ADS 验收 | | AR-3 Data Product Engineering + Governance Workbench | `planned` | Blueprint、模型、Visual/SQL/Notebook、DataOps CI/CD、质量/安全/审批共用 definition 和产品生命周期 | | AR-4 Asset/GIS Service/Spatial Experience Operations | `planned` | Service Control Plane、Features/Tiles/MVT/COG/STAC/export 及条件 legacy OGC/3D/EDR provider、Gateway/权限/缓存、原子切换/回滚、Discover/Operate/Govern 和无 LLM 多入口通过 conformance/parity/control gate | diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index 79282a5b..b0fa1128 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-31 -阶段：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-22 local authorized real-feature ingestion 已验证，权威 output promotion、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler/executor 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1 至 M3-23 local real-feature ledger promotion 已验证，可保留 output material、完整授权/成功/质量 provenance、生产观测、生产 policy/tenant isolation、生产 identity/object-store attestation、生产 consumer/scheduler/executor 和生产切换仍 `in_progress` -适用分支：`feat/ar1-metadata-fabric-real-feature-ingestion` +适用分支：`feat/ar1-metadata-fabric-real-feature-ledger-promotion` ## 判定规则 @@ -20,19 +20,19 @@ | SQL schema 历史 | PostgreSQL `schema_migrations`，以完整 migration ID + checksum 为权威 | migration CLI 的 JSON 报告 | 保持现有 ledger；任何 drift fail closed | Data Platform | AR-0，已验证 | | 部署配置策略 | Compose/K8s/进程环境；`platform_truth.CONFIG_SPECS` 定义关键类型与策略；DolphinScheduler worker 与 Active Metadata consumer 均有默认零副本、外部 ConfigMap/Secret 驱动的 Kustomize 模板和静态 validator，前者另有 staging activation preflight | `.env` 仅补默认；脱敏 snapshot、Secret key attestation、未扩容 Deployment 和 `ready_for_activation` 都是观测/模板 | 版本化 DeploymentProfile + secret reference；部署环境始终优先；模板或 preflight 通过都不等于环境已启用 | Platform/SRE/Security | AR-0，部分实现；worker 模板/preflight 本地已验证 | | 环境发布与晋级 | 本地 candidate/registry/provenance/release/live 合同已绑定 publisher、verifier、OCI 和 manifest identity；canonical `main@0182406`、archive refs、三组 active ruleset 与 `staging-provenance` protected environment 已建立，但尚无成功 publisher/verifier 或 deployment | 旧 mainline、feature branch、CI artifact、JSON、离线 report 和合成 `verified_for_staging_apply` 都不能单独成为发布权威；publisher SHA、verifier SHA 与 branch lineage 必须分别验证 | 由受保护 environment 的 DeploymentRevision 绑定 OCI、provenance artifact、release manifest 与全部 live verdict | Platform/SRE/Security/Repository Owner | AR-1 mainline 治理已恢复 -> 首次 GHCR publish/verify -> 真实 staging | -| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；Metadata Fabric recovery/metrics/policy/catalog/interoperability/failure/uncertain-commit/outbox/consumer/authorization/scheduler-delivery/projection-execution/binding-reconciliation rehearsal 均为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection/binding evidence | PlatformRun ledger 唯一登记最终状态；activation request 只拥有待授权意图；M3-19 scheduler/provider `SUCCESS`、repair/read-back 和 binding commit 仍把 Run 留在 `reconciling`，不是平台成功终局权威；本地演练进程与 evidence 不得变成生产控制器、监控后端、catalog authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 consumer、M3-16 authorization、M3-17 delivery、M3-18 projection execution 与 M3-19 binding reconciliation 本地验证；常驻受保护 executor -> staging 控制链待接入 | +| 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler managed worker 已登记但尚无生产调用方；Active Metadata consumer 登记为 `activation_request_staging_only`，其 deployment 默认为 0 replicas；Metadata Fabric recovery/metrics/policy/catalog/interoperability/failure/uncertain-commit/outbox/consumer/authorization/scheduler-delivery/projection-execution/binding-reconciliation/real-feature-ingestion/ledger-promotion rehearsal 均为 `local_verification_only` | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy/catalog/interoperability/failure/outbox/consumer/authorization/delivery/projection/binding/ingestion/promotion evidence | PlatformRun ledger 唯一登记最终状态；M3-23 临时 promotion commit 仍把 Run 留在 `accepted@0`，不是平台成功终局权威；本地演练进程、已删除 material 与 evidence 不得变成生产控制器、监控后端、catalog authority、output authority 或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker、M3-15 至 M3-23 本地控制链已验证；常驻受保护 executor、retained material 与完整 terminal evidence -> 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 | -| 数据资产身份与版本 | `gda_control.resource/resource_version` 已实现 identity、hash、predecessor、tenant FK 和幂等 gateway 写入；M3-14 新入口将新 ResourceVersion 与确定性 `resource_version.registered` 事件同事务创建，拒绝为旧版本事后补事件；`agent_data_assets`、`agent_asset_versions` 仍是兼容写路径 | UI catalog、search index、STAC、Active Metadata delivery state | GDA ledger 管身份与版本绑定；只有新原子注册路径产生权威变化事件，旧行不得伪装成实时事件；OpenMetadata 管治理目录，Gravitino 管技术对象映射 | Metadata Platform | AR-1 gateway/M3-14 本地事务已验证 -> 生产写入口切换待验收 | +| 数据资产身份与版本 | `gda_control.resource/resource_version` 已实现 identity、hash、predecessor、tenant FK 和幂等 gateway 写入；M3-14 将新 ResourceVersion 与变化事件同事务创建；M3-23 又将 output ResourceVersion 与 2 Artifacts、QualityResult、LineageEvent 作为一个 exact-replay bundle 原子追加并拒绝半状态；`agent_data_assets`、`agent_asset_versions` 仍是兼容写路径 | UI catalog、search index、STAC、Active Metadata delivery state | GDA ledger 管身份与版本绑定；authority Resource 必须预先存在，旧行不得伪装成实时事件或部分 promotion；OpenMetadata 管治理目录，Gravitino 管技术对象映射 | Metadata Platform | AR-1 gateway/M3-14/M3-23 本地事务已验证 -> retained staging/生产写入口切换待验收 | | 技术元数据 | M1 已冻结 Gravitino table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-1 固定 technical projection intent，M3-2 已在 Gravitino memory catalog 创建/read-back，M3-3 将验证后的 ref 追加到 tenant-scoped 本地 binding ledger；M3-6 又在隔离 Gravitino Basic IdP 中验证 bounded table-create、catalog-create 拒绝、密码轮换/用户吊销和完整清理；M3-7 已冻结 production identity profile/attestation gate；M3-8 已验证同一 Basic role 与 Iceberg JDBC table 在 PostgreSQL/Gravitino Pod restart 后保持；M3-9 已验证 Spark 经标准 Iceberg REST 对同一 JDBC catalog 做 read/write/schema evolution/snapshot/time travel；M3-10 已移除共享 warehouse PVC，并由跨节点 MinIO 对象检查与 Gravitino API 回读验证 Spark 结果；M3-11 已冻结 provider-neutral production object-store profile/attestation gate；M3-12 已验证 pre-forward 503 下失败提交零可见漂移、一次显式重试和无孤儿 data file | harvester 结果、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity/JDBC restart/Spark/object-store/commit-failure observation、projection plan、provider evidence、binding ledger 与 readiness report | 源系统技术对象是原始证据；Gravitino 映射并联邦，不能覆盖业务 ResourceVersion；GDA binding ledger 只记录已验证关系，本地 memory/JDBC/file-PVC/MinIO catalog evidence 不得冒充生产持久技术权威；Basic IdP、无认证 REST/HTTP、本地主机对象存储、pending profile 和合成 attestation 都不是生产身份或生产 storage，M3-11 合同不构成 provider selection/deployment，M3-12 本地证据不构成网络 exactly-once 或生产 reconcile | Metadata Platform | AR-1 M1/M2 + M3-6 identity + M3-7 gate + M3-8 local persistence + M3-9/M3-10 interoperability + M3-11 object-store gate + M3-12 commit-failure recovery 已验证 -> 受保护身份/生产对象存储 attestation/uncertain outcome reconcile/完整 Spark-Flink conformance 待执行 | | 治理目录 | M1 已冻结 OpenMetadata table ref/reconciliation；M2 已验证本地 foundation/recovery/metrics/policy 和 production readiness contracts；M3-2 已用 bootstrap admin 创建目标并回读真实 UUID；M3-3 将该 UUID 经 evidence gate 追加到本地 GDA binding ledger；M3-4 将精确 OpenLineage candidate 经 outbox 投递到本地 HTTP receiver；M3-5 已验证临时非管理员 bot 的 scoped `table/Create` grant、policy-create 拒绝及 JWT 轮换/吊销；M3-7 将其 allow/deny 范围纳入双 provider production identity gate | 搜索/页面视图、合成 response、本地 sandbox/recovery/metrics/policy/ingestion/identity observation、projection/provider evidence、binding ledger、lineage outbox/receipt、OpenLineage event 与 readiness report | OpenMetadata 为 owner/glossary/classification/quality discoverability 权威；GDA ledger 保留审批/provider identity，outbox 只拥有投递状态，receiver 拥有接收状态；pending identity profile 和合成 attestation 均不反写 ResourceVersion 或建立生产权威 | Governance | AR-1 M1/M2 + M3-5 local bounded identity + M3-7 readiness contract 已验证 -> protected identity ingestion/生产持久 binding/受保护 production receiver 待执行 | -| 血缘 | `gda_control.lineage_event` 已实现 immutable version edge 和幂等 gateway ingest；`agent_asset_lineage` 旧记录仍是可变 asset edge | OpenMetadata lineage graph、UI DAG | 只有 source/target ResourceVersion 与 event checksum 证据完整的旧记录可形成 eligible plan；目录图只作可重建投影 | Data Platform | AR-1 gateway 已验证 -> adapter 待接入 | +| 血缘 | `gda_control.lineage_event` 已实现 immutable version edge 和幂等 gateway ingest；M3-23 已将真实 source/output version edge 与 output Artifact 在同一 promotion 事务写入；`agent_asset_lineage` 旧记录仍是可变 asset edge | OpenMetadata lineage graph、UI DAG | 只有 source/target ResourceVersion、Run input、Definition、Artifact 与 event checksum 完整匹配才可形成权威 edge；目录图只作可重建投影 | Data Platform | AR-1 gateway/M3-23 本地 PostgreSQL 已验证 -> staging adapter 待接入 | | 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/生产切换待验收 | +| Run 最终状态 | `gda_control.platform_run/event` 已实现受控 submit/read/CAS；通用 transition 已禁止 `succeeded`，专用数据库 finalizer 只接受精确 workload、success observation、内容匹配 output、独立 passed QualityResult/evidence 和 input-to-output lineage；M3-23 即使五类 output facts 已落账，显式 finalization 仍被拒绝并保持 `accepted@0` | Redis progress、日志、DolphinScheduler state、attempt observation | 旧 run 到 PlatformRun 永久 prohibited；provider/ledger facts 只形成 observation 与 reconciliation 输入，数据库 evidence gate 唯一裁决成功 | DataOps/AgentOps | AR-1 success authority/M3-23 negative gate 本地 PostgreSQL 已验证 -> 完整 staging terminal evidence 待验收 | | 调度与补数 | APScheduler、自进化 scheduler 和调用方定时逻辑并存；DolphinScheduler POC 只验证 manual start/list/variables/STOP | UI schedule 列表 | DolphinScheduler 管 DataOps schedule/complement；Temporal 只管需要 durable signal/compensation 的 Agent/GWM workflow | DataOps/AgentOps | AR-1 manual correlation 已验证；schedule/complement/failover 待验收 | | 事件交付 | Standards outbox 已数据库耐久；`platform_command_outbox` 支持 DolphinScheduler dispatch/reconcile；M3-14 `metadata_change_outbox` 将新 ResourceVersion 与内容绑定事件同事务写入；M3-15 managed consumer 同事务创建 inert request；M3-16 将真实 ResourceVersion、Definition/Run/plan/PolicyDecision/Approval/authorizer 绑定后与一个 pending dispatch 同事务提交；M3-17 由既有 consumer 向本地真实 DolphinScheduler 提交并回读；M3-18 的单个任务触发独立授权的 provider apply/read-back 与零写 replay；M3-19 在 exact OpenMetadata 前提下修复缺失 Gravitino projection、零写 replay 后提交 immutable binding | command/metadata delivery status、消费者 claim、activation intent/request/authorization、FrameworkAttemptObservation、provider instance/correlation、provider apply/read-back/binding、worker status JSON、WebSocket 消息 | command/event 与源事实同事务入 outbox，幂等 consumer 交付；Active Metadata consumer 不能授权或执行；dispatch 与 provider apply 分别授权；scheduler/provider `SUCCESS` 和 binding persistence 仍须经平台终局 evidence gate | Platform/Integrations/Metadata Platform | AR-1 command worker、M3-14 至 M3-19 本地已验证 -> protected authorizer/worker/executor identity、生产 scheduler/provider 和 production scale-up 待执行 | -| 质量结果 | `gda_control.quality_result` 已提供 tenant RLS、append-only gateway 写入，绑定 Run、output ResourceVersion、rule version、verdict、metrics、evidence Artifact 和独立 evaluator；standards、QC、MMFE 专项结果仍未迁移 | dashboard、OpenMetadata quality summary | GDA ledger 保存产品终局所需的不可变 verdict/evidence；OpenMetadata 与 UI 只作可重建发现投影；旧结果缺稳定版本和证据时不得升级为终局依据 | Governance/DataOps | AR-1 最小成功证据已验证 -> 真实规则/staging 待接入 | +| 质量结果 | `gda_control.quality_result` 已提供 tenant RLS、append-only gateway 写入，绑定 Run、output ResourceVersion、rule version、verdict、metrics、evidence Artifact 和独立 evaluator；M3-23 已原子写入真实 20-feature 六项 passed metrics，但证据 Artifact creator 仍不是独立 evaluator；standards、QC、MMFE 专项结果仍未迁移 | dashboard、OpenMetadata quality summary | GDA ledger 保存产品终局所需的不可变 verdict/evidence；只有独立 evaluator 生成并绑定的 evidence 才满足 success gate；OpenMetadata 与 UI 只作可重建投影 | Governance/DataOps | AR-1/M3-23 ledger commit 已验证 -> 独立 quality evidence provenance/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 待执行 | | GIS 服务定义与 active revision | Martin、REST/MVT/STAC endpoints 和配置直接暴露 | Ingress、tile cache、客户端图层 | GIS Service Control Plane 管 Service/Layer/Style/TMS/DeploymentRevision；provider/Gateway 仅执行 | GIS Platform | AR-4 | @@ -66,6 +66,7 @@ 21. M3-20 不修改 M3-19 binding schema、ledger 或 evidence，而是为同一重庆 ResourceVersion 新建 runtime-bound durable promotion candidate。受限 Basic principal 在隔离 JDBC metadata + warehouse PVC target 中只创建一次表；即时 replay 与 PostgreSQL/Gravitino restart 后第一次 replay 均为 `no_op/0 mutations`。cluster/namespace/Service/StatefulSet/PVC/image identity 被绑定且重启前后稳定，Pod UID 必须变化；candidate 未写 GDA Control，namespace/PVC 已清理。这只证明本地 restart continuity，不证明生产 durable catalog/object store、protected identity、OIDC/TLS、生产 ingestion 或 readiness。 22. M3-21 不修改 M3-20/M3-19 历史，而是以 M3-20 candidate 为 predecessor，将同一重庆 ResourceVersion 投影到 JDBC catalog + 跨节点 MinIO warehouse。稳定 binding 包含双 Service、三个 StatefulSet、PostgreSQL/MinIO PVC、镜像、节点和 S3 配置，Gravitino 无 warehouse PVC。首次 apply 为 1 个 table create，即时及有序重启后首个 replay 均为 `no_op/0`；直接 S3 metadata key/ETag/body SHA/表 schema 不变。该表没有 source feature rows，candidate 未落账，所有临时资源已清理。这不证明生产对象存储、durable catalog、protected identity、TLS/OIDC、生产 ingestion 或 readiness。 23. M3-22 以 M3-21 candidate 为 predecessor，将同一重庆 bundle 的 20 个真实 EPSG:4490 feature rows 规范化为八列、由精确 PolicyDecision/Approval 授权 Spark/Sedona 写入 JDBC/S3 Iceberg。六项质量计数均为 20，首次执行 `appended/1` 且只有 1 个 snapshot/Parquet，即时 replay 为 `no_op/0`；S3 直读为 1 data + 2 metadata + 2 manifest。输出 ResourceVersion、Artifact、passed QualityResult 与 LineageEvent 只是 path-free candidates，未写 GDA Control，Run 未成功终局；namespace/PV/port-forward 已清理。这不证明生产对象存储、protected identity、完整 engine conformance、生产 ingestion 或 readiness。 +24. M3-23 只证明 M3-22 path-free candidates 可在临时 PostgreSQL GDA Control 中按 `ResourceVersion -> output Artifact -> quality evidence Artifact -> QualityResult -> LineageEvent` 单事务追加。缺 authority、半状态、跨租户和 direct mutation 均拒绝，故障注入整笔回滚，精确 replay 不新增；Run 保持 `accepted@0` 且 success finalization 被拒绝。M3-22 material、M3-23 数据库与完整 authorization Artifacts 均未保留或伪造补写；这不证明 persistent authority、terminal success、staging/生产 ingestion 或 readiness。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -110,11 +111,13 @@ - Metadata Fabric M3-19 已在 retained OpenMetadata snapshot 精确匹配 M3-18 后，识别 Gravitino `memory` connector 空状态与持久 entity index 残留。专用 catalog 配置精确且 visible schema inventory 为零后，scheduler callback 只执行 `gravitino.catalog.reset_stale_empty_memory`、catalog/schema/table create 四次 mutation；OpenMetadata 零写入，立即 replay 为 `no_op/0 mutations`，binding SHA 保持 `7de24cee9dd50dfeefcc886cf43024f4d92b7650767d71d064fdce19ffccb16b`。PlatformGateway 首次 commit `created=true`、重放 `created=false`、仅 1 行，FORCE RLS、跨租户隔离、append-only 和 direct UPDATE/DELETE 拒绝通过；Run 保持 `reconciling`，临时资源均清理。contract fingerprint 为 `012a7c86ba9fe53217e721ff7286b8f2a246b9394efd2999abbcd025e13ac7f5`，evidence fingerprint 为 `e6d0e3ac4e052029dad0c18d0804626a8af61554a54081c37d8cc9a80c55cd33`。`durable_catalog_verified=false`、`production_ingestion_verified=false`、`production_ready=false`。 - Metadata Fabric M3-20 已将同一重庆 ResourceVersion 的 retained OpenMetadata ref 与新 Gravitino JDBC target 组合为 logical binding，并另行绑定 Docker Desktop cluster UID、namespace/Service/StatefulSet/PVC/image identity。受限 `gda-cultural-district-projector` 只有 `USE_CATALOG`、`USE_SCHEMA`、`CREATE_TABLE`，catalog create 前后均为 403；首次 projection 为 `created/1 gravitino.table.create`，即时 replay 与 PostgreSQL -> Gravitino restart 后第一次 replay 均为 `no_op/0 mutations`，表 projection/snapshot 不变。logical/runtime/promotion SHA 分别为 `8c312db37bfe92e034bcdcb7a3c35847c81e862c74a3437970def1007af42750`、`a78975311fc34abd76fa41dea581594806b3d18ed364ba518cfc44c4204822f7`、`bb6672cb7f98fa53305e17bbca2cb5b3756d4a335a94d79114fb4184273871d1`；contract/evidence SHA 为 `307f2d4390028589c0f38be859c53826bd149d7f2a133b14488230d4f5ff6eb8` / `53773e9417668e03ad3ab2b5c3cdbd627fb3bc397d63c5860755ec5318eebe8b`。candidate 未落 ledger，namespace/PVC/port-forward 已清理；`durable_catalog_verified=false`、`production_object_store_verified=false`、`production_ready=false`。 - Metadata Fabric M3-21 已把 M3-20 candidate 作为 predecessor，在 M3-10 JDBC + 跨节点 MinIO runtime 中创建同一重庆 ResourceVersion 的独立 target。受限 `gda-object-store-cultural-district-projector` 首次只创建一次表，catalog create 前后均为 403；即时和 PostgreSQL -> Gravitino restart 后首个 replay 均为 `no_op/0`。Gravitino 无 warehouse PVC；MinIO 位于 `desktop-control-plane`，provider 位于 `desktop-worker`。直接 S3 检查在 `warehouse/cultural_heritage/cultural_districts/` 下只找到 1 个 metadata JSON，无 data/manifest，key/ETag/body SHA/表 schema 重启前后相同。predecessor/logical/runtime/promotion SHA 分别为 `bb6672cb7f98fa53305e17bbca2cb5b3756d4a335a94d79114fb4184273871d1`、`614ce5e4c45dba1437dc888cbd79b2d58954184113a62c20170ab84b5570d9e1`、`dd63917b6354a2e92853763ddc3e3a981cb40717f84c0f819b1a4e6844ae100b`、`63812c311b3f239bc6a944748c4ff384250eb9c9ed9009d3384fc699f1d3eaa9`；contract/evidence SHA 为 `b1a2db34a70eaa7dd55da1d6c85da9f420c755c71868aafe7972e3794034a6cc` / `d73754c53cf16d888aa345baa5d079cc7fd98d8b84db747f52188c1a69bf1628`。candidate 未落 ledger、feature rows 未 ingest、临时资源已清理；生产对象存储与 `production_ready` 仍为 `false`。 +- Metadata Fabric M3-22 已由受授权 Spark `3.5.0` + Sedona `1.9.0` 将同一重庆 bundle 的 20 个 EPSG:4490 features 写入 JDBC/S3 Iceberg。六项空间质量计数均为 20，首次 `appended/1`、1 snapshot/Parquet，即时 replay `no_op/0`；S3 直读为 1 data + 2 metadata + 2 manifest。row-set/output/contract/evidence SHA 为 `c26ff708f4b6be082327dff63a6a8659420dbc4cab37dea1cac7b40f147512df` / `bdc06792e8b935176ee6df6f6f6d4be1535622d54d9b994a778cabfe5a574618` / `af211f2d2f4830decb9ffe369cd9e7ec2c9349c2e2c8bd789347a6fdc288e1dc` / `42abd82613eaf28cb53c64280258bc75dba6cf841f9a513a4c801a9f798b9899`。output/quality/lineage 仍是 candidates，runtime/material 已清理，Run 未终局。 +- Metadata Fabric M3-23 已把 M3-22 的 output ResourceVersion、output Artifact、quality evidence Artifact、独立 passed QualityResult 与 source-to-output LineageEvent 作为一个 `RunOutputLedgerPromotion`，由独立 promoter 在一个 PlatformGateway 事务中写入真实临时 PostgreSQL，同时保持此前 gateway-bound evidence 指纹不变。缺 authority 先验拒绝；QualityResult 前故障注入后候选计数全为 0；首次 `created=true`、精确 replay `created=false`，最终计数为 `1/2/1/1`。FORCE RLS、最小 grant、跨租户读/直写和八个 direct UPDATE/DELETE 拒绝通过；Run 保持 `accepted@0`，success finalization 被既有 gate 拒绝。promotion/contract/evidence SHA 为 `404b6e4e5d8194f092bd83ef99cbf2d1d727015b926cd438a79eb0210f969a22` / `bd21c81925f66acdfecca5cabd78651f31deab4165da2ccd6900c4e5796e5735` / `f6efea5000791dec1716a8354a8e39a8425b083ca4d409f4bcb61f0e7e03580d`。M3-22 material 与临时数据库已清理，完整 authorization Artifacts 未补写；`output_material_retained=false`、`platform_run_succeeded=false`、`production_ready=false`。 ## 下一验收证据 -- M3-21 的本地 MinIO 空表 metadata promotion 不计入生产对象存储或 ingestion 退出门；其真实重庆 feature slice 后续由 M3-22 独立验收，历史 candidate 与 evidence 保持不变； -- M3-22 的本地真实 feature slice 已通过受授权 Spark/Sedona、单 snapshot/no-op replay 与直接 S3 readback，但不计入生产 ingestion 退出门；下一步是将 output/quality/lineage candidates 原子晋级到 GDA Control，并继续把 Run 终局、生产 provider attestation 与 staging-scale ingestion 独立验收； +- M3-21 空表 metadata promotion、M3-22 临时真实 feature ingestion 与 M3-23 临时 ledger promotion 均不计入生产对象存储、持久 authority 或 ingestion 退出门，三阶段历史 candidate/evidence 保持不变； +- M3-23 已通过 M3-22 candidates 的临时 GDA Control 原子晋级、失败回滚、精确 replay 与 security negative gates，但不计入持久/生产 authority；下一步是在可保留 staging material 上持久化完整 PolicyDecision/Approval、provider success observation 和由独立 evaluator 创建的 quality evidence，再经同一 promotion 与既有 success finalizer 完成非临时 Run 终局； - 完成首次 application subject publish 与 protected verifier run；当前 mainline、archive refs、ruleset、required reviewer、禁止 bypass 和 environment enable variable 已配置并复核； - 真实 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 运行产物； diff --git a/scripts/metadata-fabric-real-feature-ledger-promotion.sh b/scripts/metadata-fabric-real-feature-ledger-promotion.sh new file mode 100755 index 00000000..332581c3 --- /dev/null +++ b/scripts/metadata-fabric-real-feature-ledger-promotion.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_real_feature_ledger_promotion "$@"