feat!: enforce single-path completeness operations

Author: Cataldir

CHANGELOG.md

@@ -7,6 +7,38 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.0.0] - 2026-03-04
+
+> **Release**: `v2.0.0`
+> **Theme**: Single-path completeness operations (breaking compatibility removal)
+
+### Changed
+
+- Product consistency validation service now uses a **single canonical operation path**:
+  - schema-driven completeness evaluation only
+  - no legacy 4-field validator path
+  - no legacy `product-events` compatibility subscription
+
+### Removed
+
+- Legacy compatibility components removed from `product-management-consistency-validation`:
+  - legacy validator logic (`missing_name`, `negative_price`, `missing_currency`, `missing_image`)
+  - legacy event processing module for `product-events`
+  - legacy MCP tools:
+    - `/product/consistency/check`
+    - `/product/consistency/product`
+
+### Added
+
+- Canonical MCP operation for completeness:
+  - `/product/completeness/evaluate`
+
+### Migration Notes
+
+- Any integration relying on the removed consistency MCP tools must switch to `/product/completeness/evaluate`.
+- Event-driven completeness processing now requires publishing to `completeness-jobs` (consumer group `completeness-engine`).
+- This release intentionally removes backward compatibility to guarantee a single operation model.
+
 ## [1.1.0] - 2026-03-03
 
 > **Release**: [v1.1.0](https://github.com/Azure-Samples/holiday-peak-hub/releases/tag/v1.1.0)

apps/README.md

@@ -67,7 +67,7 @@ Every agent service follows an identical pattern:
 |-----|-------------|
 | [product-management-acp-transformation](product-management-acp-transformation/) | Transforms catalog products into ACP-compliant (Agentic Commerce Protocol) payloads, ensuring all required fields are populated and flagging missing data for remediation. Subscribes to `product-events`. |
 | [product-management-assortment-optimization](product-management-assortment-optimization/) | Ranks products by performance indicators (rating, price, demand) and recommends optimal assortment sets of a target size, explaining keep/drop trade-offs. Subscribes to `order-events` and `product-events`. |
-| [product-management-consistency-validation](product-management-consistency-validation/) | Validates catalog product consistency — checks field completeness, detects data quality issues, and provides remediation steps for invalid items. Subscribes to `product-events`. |
+| [product-management-consistency-validation](product-management-consistency-validation/) | Evaluates schema-driven product completeness with weighted scoring, gap reports, and enrichment triggers. Uses a single completeness operation model across `/invoke`, MCP, and events. Subscribes to `completeness-jobs`. |
 | [product-management-normalization-classification](product-management-normalization-classification/) | Normalizes product names, categories, and tags into a canonical taxonomy, assigns classifications, and highlights missing attributes that need enrichment. Subscribes to `product-events`. |
 
 ## Infrastructure

apps/product-management-consistency-validation/src/product_management_consistency_validation/adapters.py

@@ -7,7 +7,6 @@
 
 from holiday_peak_lib.adapters.mock_adapters import MockProductAdapter
 from holiday_peak_lib.adapters.product_adapter import ProductConnector
-from holiday_peak_lib.schemas.product import CatalogProduct
 
 from .completeness_engine import CategorySchema, GapReport
 
@@ -71,42 +70,18 @@ async def store_gap_report(self, report: GapReport) -> None:
 
 @dataclass
 class ProductConsistencyAdapters:
-    """Container for product consistency validation adapters."""
+    """Container for completeness evaluation adapters."""
 
     products: ProductConnector
-    validator: "ProductConsistencyValidator"
     completeness: CompletenessStorageAdapter = field(default_factory=CompletenessStorageAdapter)
 
 
-class ProductConsistencyValidator:
-    """Validate product data for completeness and consistency."""
-
-    async def validate(self, product: CatalogProduct) -> dict[str, Any]:
-        issues = []
-        if not product.name:
-            issues.append("missing_name")
-        if product.price is not None and product.price < 0:
-            issues.append("negative_price")
-        if product.price is not None and not product.currency:
-            issues.append("missing_currency")
-        if not product.image_url:
-            issues.append("missing_image")
-        return {
-            "sku": product.sku,
-            "issues": issues,
-            "status": "invalid" if issues else "valid",
-        }
-
-
 def build_consistency_adapters(
     *,
     product_connector: Optional[ProductConnector] = None,
     completeness_adapter: Optional[CompletenessStorageAdapter] = None,
 ) -> ProductConsistencyAdapters:
-    """Create adapters for product consistency validation workflows."""
+    """Create adapters for schema-driven completeness workflows."""
     products = product_connector or ProductConnector(adapter=MockProductAdapter())
-    validator = ProductConsistencyValidator()
     completeness = completeness_adapter or CompletenessStorageAdapter()
-    return ProductConsistencyAdapters(
-        products=products, validator=validator, completeness=completeness
-    )
+    return ProductConsistencyAdapters(products=products, completeness=completeness)

apps/product-management-consistency-validation/src/product_management_consistency_validation/agents.py

@@ -6,83 +6,119 @@
 from typing import Any
 
 from holiday_peak_lib.adapters import BaseCRUDAdapter
-from holiday_peak_lib.adapters.acp_mapper import AcpCatalogMapper
 from holiday_peak_lib.agents import BaseRetailAgent
 from holiday_peak_lib.agents.fastapi_mcp import FastAPIMCPServer
 
 from .adapters import ProductConsistencyAdapters, build_consistency_adapters
+from .completeness_engine import CompletenessEngine
 
 
 class ProductConsistencyAgent(BaseRetailAgent):
-    """Agent that validates catalog product consistency."""
+    """Agent that evaluates products with the schema-driven completeness engine."""
 
     def __init__(self, config, *args: Any, **kwargs: Any) -> None:
         super().__init__(config, *args, **kwargs)
         self._adapters = build_consistency_adapters()
+        self._engine = CompletenessEngine()
 
     @property
     def adapters(self) -> ProductConsistencyAdapters:
         return self._adapters
 
+    async def evaluate_completeness(
+        self, sku: str, category_id: str | None = None
+    ) -> dict[str, Any]:
+        product = await self.adapters.products.get_product(str(sku))
+        if not product:
+            return {"error": "sku not found", "sku": sku}
+
+        resolved_category = category_id or product.category or "default"
+        schema = await self.adapters.completeness.get_schema(resolved_category)
+        if schema is None:
+            return {
+                "error": "schema not found",
+                "sku": sku,
+                "category_id": resolved_category,
+            }
+
+        report = self._engine.evaluate(str(sku), product.model_dump(), schema)
+        await self.adapters.completeness.store_gap_report(report)
+
+        return {
+            "sku": sku,
+            "category_id": resolved_category,
+            "completeness": report.model_dump(mode="json"),
+            "needs_enrichment": bool(report.enrichable_gaps),
+        }
+
     async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
         sku = request.get("sku")
         if not sku:
             return {"error": "sku is required"}
-        product = await self.adapters.products.get_product(str(sku))
-        if not product:
-            return {"error": "sku not found", "sku": sku}
 
-        validation = await self.adapters.validator.validate(product)
-        acp_product = AcpCatalogMapper().to_acp_product(product, availability="unknown")
+        result = await self.evaluate_completeness(
+            sku=str(sku), category_id=request.get("category_id")
+        )
+
+        if "error" in result:
+            return result
 
         if self.slm or self.llm:
             messages = [
                 {"role": "system", "content": _consistency_instructions()},
                 {
                     "role": "user",
-                    "content": {
-                        "sku": sku,
-                        "product": product.model_dump(),
-                        "acp_product": acp_product,
-                        "validation": validation,
-                    },
+                    "content": result,
                 },
             ]
             return await self.invoke_model(request=request, messages=messages)
 
-        return {
-            "service": self.service_name,
-            "sku": sku,
-            "product": product.model_dump(),
-            "acp_product": acp_product,
-            "validation": validation,
-        }
+        return {"service": self.service_name, **result}
 
 
 def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
-    """Expose MCP tools for product consistency validation workflows."""
-    adapters = getattr(agent, "adapters", build_consistency_adapters())
+    """Expose MCP tools for schema-driven completeness workflows."""
+    completeness_agent = agent if isinstance(agent, ProductConsistencyAgent) else None
+    adapters = (
+        completeness_agent.adapters
+        if completeness_agent is not None
+        else build_consistency_adapters()
+    )
+    engine = CompletenessEngine()
 
-    async def validate_product(payload: dict[str, Any]) -> dict[str, Any]:
+    async def evaluate_product_completeness(payload: dict[str, Any]) -> dict[str, Any]:
         sku = payload.get("sku")
         if not sku:
             return {"error": "sku is required"}
+        category_id = payload.get("category_id")
+        if completeness_agent is not None:
+            return await completeness_agent.evaluate_completeness(
+                sku=str(sku), category_id=category_id
+            )
+
         product = await adapters.products.get_product(str(sku))
         if not product:
             return {"error": "sku not found", "sku": sku}
-        validation = await adapters.validator.validate(product)
-        acp_product = AcpCatalogMapper().to_acp_product(product, availability="unknown")
-        return {"validation": validation, "acp_product": acp_product}
 
-    async def get_product(payload: dict[str, Any]) -> dict[str, Any]:
-        sku = payload.get("sku")
-        if not sku:
-            return {"error": "sku is required"}
-        product = await adapters.products.get_product(str(sku))
-        return {"product": product.model_dump() if product else None}
+        resolved_category = category_id or product.category or "default"
+        schema = await adapters.completeness.get_schema(resolved_category)
+        if schema is None:
+            return {
+                "error": "schema not found",
+                "sku": sku,
+                "category_id": resolved_category,
+            }
+
+        report = engine.evaluate(str(sku), product.model_dump(), schema)
+        await adapters.completeness.store_gap_report(report)
+        return {
+            "sku": str(sku),
+            "category_id": resolved_category,
+            "completeness": report.model_dump(mode="json"),
+            "needs_enrichment": bool(report.enrichable_gaps),
+        }
 
-    mcp.add_tool("/product/consistency/check", validate_product)
-    mcp.add_tool("/product/consistency/product", get_product)
+    mcp.add_tool("/product/completeness/evaluate", evaluate_product_completeness)
     _register_crud_tools(mcp)
 
 
@@ -95,7 +131,7 @@ def _register_crud_tools(mcp: FastAPIMCPServer) -> None:
 
 def _consistency_instructions() -> str:
     return (
-        "You are a product consistency validation agent. "
-        "Check catalog completeness and highlight missing fields. "
-        "Provide remediation steps for invalid items."
+        "You are a schema-driven completeness engine agent. "
+        "Evaluate products against category schemas and return weighted completeness scores. "
+        "Highlight missing or invalid fields and enrichment candidates."
     )

apps/product-management-consistency-validation/src/product_management_consistency_validation/event_handlers.py

@@ -14,9 +14,6 @@
 from product_management_consistency_validation.event_consumer import (
     build_completeness_event_handlers,
 )
-from product_management_consistency_validation.event_handlers import (
-    build_event_handlers,
-)
 
 SERVICE_NAME = "product-management-consistency-validation"
 memory_settings = MemorySettings()
@@ -52,13 +49,9 @@
     else None
 )
 
-# Merge product-events handlers (backward compat) with completeness-jobs handlers
-_all_handlers = {
-    **build_event_handlers(),
-    **build_completeness_event_handlers(
-        completeness_threshold=float(os.getenv("COMPLETENESS_THRESHOLD", "0.7"))
-    ),
-}
+_all_handlers = build_completeness_event_handlers(
+    completeness_threshold=float(os.getenv("COMPLETENESS_THRESHOLD", "0.7"))
+)
 
 app = build_service_app(
     SERVICE_NAME,
@@ -87,7 +80,6 @@
     lifespan=create_eventhub_lifespan(
         service_name=SERVICE_NAME,
         subscriptions=[
-            EventHubSubscription("product-events", "validation-group"),
             EventHubSubscription("completeness-jobs", "completeness-engine"),
         ],
         handlers=_all_handlers,

apps/product-management-consistency-validation/src/product_management_consistency_validation/main.py

@@ -1,6 +1,6 @@
 [project]
 name = "product-management-consistency-validation"
-version = "0.1.0"
+version = "1.0.0"
 description = "Product consistency validation"
 authors = [{name = "Ricardo Cataldi", email = "rcataldi@microsoft.com"}]
 requires-python = ">=3.13"