feat(platform): add durable active metadata consumer

.github/workflows/ci.yml

@@ -30,6 +30,7 @@ on:
       - feat/ar1-metadata-fabric-spark-commit-failure-recovery
       - feat/ar1-metadata-fabric-spark-uncertain-commit-reconciliation
       - feat/ar1-metadata-fabric-active-metadata-outbox
+      - feat/ar1-metadata-fabric-active-metadata-consumer
 
 env:
   PYTHON_VERSION: "3.13"
@@ -169,6 +170,12 @@ jobs:
       - name: Validate metadata fabric Active Metadata outbox evidence
         run: python -m data_agent.metadata_fabric_active_metadata_outbox validate
 
+      - name: Validate metadata fabric Active Metadata consumer evidence
+        run: python -m data_agent.metadata_fabric_active_metadata_consumer validate
+
+      - name: Validate Active Metadata consumer deployment boundary
+        run: python -m data_agent.active_metadata_consumer_deployment validate
+
       - name: Validate DolphinScheduler adapter boundary
         run: python -m data_agent.dolphinscheduler_adapter validate
 
@@ -200,6 +207,11 @@ jobs:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test
         run: python -m pytest data_agent/test_metadata_fabric_active_metadata_outbox_postgres.py -q
 
+      - name: Verify Active Metadata consumer on PostgreSQL
+        env:
+          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test
+        run: python -m pytest data_agent/test_active_metadata_consumer_postgres.py -q
+
       - name: Run required platform tests
         env:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/gis_agent_test
@@ -229,7 +241,11 @@ jobs:
             data_agent/test_metadata_fabric_ingestion_replay.py \
             data_agent/test_metadata_fabric_binding_contract.py \
             data_agent/test_active_metadata_change_contract.py \
+            data_agent/test_active_metadata_consumer.py \
+            data_agent/test_active_metadata_consumer_deployment.py \
+            data_agent/test_active_metadata_consumer_worker.py \
             data_agent/test_metadata_fabric_active_metadata_outbox.py \
+            data_agent/test_metadata_fabric_active_metadata_consumer.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 \

data_agent/active_metadata_change_contract.py

@@ -29,6 +29,7 @@
 DELIVERY_SCHEMA = "gda.metadata_change_delivery.v1"
 REGISTRATION_SCHEMA = "gda.active_metadata_registration.v1"
 ACTIVATION_INTENT_SCHEMA = "gda.metadata_activation_intent.v1"
+ACTIVATION_REQUEST_SCHEMA = "gda.metadata_activation_request.v1"
 RESOURCE_VERSION_REGISTERED = "resource_version.registered"
 METADATA_PROJECTION_ROUTE = "metadata_fabric.projection_plan"
 
@@ -284,6 +285,64 @@ def build_metadata_activation_intent(
     )
 
 
+class MetadataActivationRequest(_FrozenModel):
+    request_schema: Literal["gda.metadata_activation_request.v1"] = Field(
+        default=ACTIVATION_REQUEST_SCHEMA,
+        alias="schema",
+    )
+    request_id: UUID
+    intent: MetadataActivationIntent
+    status: Literal["awaiting_authorization"] = "awaiting_authorization"
+    provider_apply_authorized: Literal[False] = False
+    provider_mutations_executed: Literal[False] = False
+    production_scheduler_submission_verified: Literal[False] = False
+    production_ingestion_verified: Literal[False] = False
+    production_ready: Literal[False] = False
+    request_sha256: Sha256
+
+    @model_validator(mode="after")
+    def _content_bound(self) -> Self:
+        expected_id = uuid5(
+            self.intent.event_id,
+            f"metadata-activation-request:{self.intent.intent_sha256}",
+        )
+        if self.request_id != expected_id:
+            raise ValueError("MetadataActivationRequest ID does not match its intent")
+        stable = self.model_dump(
+            mode="json",
+            by_alias=True,
+            exclude={"request_sha256"},
+        )
+        if self.request_sha256 != canonical_json_fingerprint(stable):
+            raise ValueError("MetadataActivationRequest SHA-256 does not match")
+        return self
+
+
+def build_metadata_activation_request(
+    intent: MetadataActivationIntent,
+) -> MetadataActivationRequest:
+    request_id = uuid5(
+        intent.event_id,
+        f"metadata-activation-request:{intent.intent_sha256}",
+    )
+    stable = {
+        "schema": ACTIVATION_REQUEST_SCHEMA,
+        "request_id": str(request_id),
+        "intent": intent.model_dump(mode="json", by_alias=True),
+        "status": "awaiting_authorization",
+        "provider_apply_authorized": False,
+        "provider_mutations_executed": False,
+        "production_scheduler_submission_verified": False,
+        "production_ingestion_verified": False,
+        "production_ready": False,
+    }
+    return MetadataActivationRequest(
+        request_id=request_id,
+        intent=intent,
+        request_sha256=canonical_json_fingerprint(stable),
+    )
+
+
 class MetadataChangeDelivery(_FrozenModel):
     delivery_schema: Literal["gda.metadata_change_delivery.v1"] = Field(
         default=DELIVERY_SCHEMA,

data_agent/active_metadata_consumer.py

@@ -0,0 +1,98 @@
+"""Thin consumer that durably stages Active Metadata activation requests."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import UUID
+
+from .active_metadata_change_contract import (
+    build_metadata_activation_intent,
+    build_metadata_activation_request,
+)
+from .platform_gateway import (
+    GatewayConflictError,
+    GatewayValidationError,
+    PlatformGateway,
+)
+
+
+@dataclass(frozen=True)
+class ActiveMetadataBatchResult:
+    claimed: int
+    staged: int
+    replayed: int
+    retry_pending: int
+    failed: int
+    request_ids: tuple[UUID, ...]
+
+
+class ActiveMetadataConsumer:
+    """Route changes to inert requests without owning authorization or execution."""
+
+    def __init__(self, gateway: PlatformGateway, *, consumer_subject: str):
+        if not consumer_subject.startswith("workload:"):
+            raise ValueError("Active Metadata consumer must use workload identity")
+        self.gateway = gateway
+        self.consumer_subject = consumer_subject
+
+    def run_once(
+        self,
+        tenant_id: str,
+        *,
+        worker_id: str,
+        limit: int = 10,
+        lease_seconds: int = 60,
+    ) -> ActiveMetadataBatchResult:
+        deliveries = self.gateway.claim_metadata_changes(
+            tenant_id,
+            worker_id,
+            consumer_subject=self.consumer_subject,
+            limit=limit,
+            lease_seconds=lease_seconds,
+        )
+        staged = 0
+        replayed = 0
+        retry_pending = 0
+        failed = 0
+        request_ids: list[UUID] = []
+        for delivery in deliveries:
+            intent = build_metadata_activation_intent(
+                delivery.event,
+                routed_by=self.consumer_subject,
+            )
+            request = build_metadata_activation_request(intent)
+            try:
+                result = self.gateway.stage_metadata_activation_request(
+                    tenant_id,
+                    delivery.event.event_id,
+                    worker_id=worker_id,
+                    request=request,
+                )
+            except GatewayConflictError:
+                # The commit outcome or lease owner is uncertain. Leave the
+                # claim for deterministic reclaim instead of declaring failure.
+                retry_pending += 1
+                continue
+            except GatewayValidationError:
+                self.gateway.fail_metadata_change(
+                    tenant_id,
+                    delivery.event.event_id,
+                    worker_id=worker_id,
+                    error_code="activation_contract_rejected",
+                    retryable=False,
+                )
+                failed += 1
+                continue
+            request_ids.append(request.request_id)
+            if result.created:
+                staged += 1
+            else:
+                replayed += 1
+        return ActiveMetadataBatchResult(
+            claimed=len(deliveries),
+            staged=staged,
+            replayed=replayed,
+            retry_pending=retry_pending,
+            failed=failed,
+            request_ids=tuple(request_ids),
+        )