PIM writeback on HITL approval (#390)

* truth-hitl: support enhanced review payload shapes for UI

* Add HITL approval-driven PIM writeback flow

* Sort imports for truth HITL/export modules

* Format truth HITL/export files for lint

Author: Cataldir

apps/truth-export/src/truth_export/event_handlers.py

@@ -39,16 +39,25 @@ async def handle_export_job(partition_context, event) -> None:  # noqa: ANN001
             return
 
         if _is_hitl_writeback_event(payload, data, protocol):
-            result = await engine.writeback_entity(
+            approved_fields_raw = data.get("approved_fields")
+            approved_fields = (
+                [str(field) for field in approved_fields_raw if field]
+                if isinstance(approved_fields_raw, list)
+                else None
+            )
+
+            result = await engine.writeback_to_pim(
                 adapters.writeback_manager,
+                adapters.truth_store,
                 str(entity_id),
+                approved_attributes=approved_fields,
                 dry_run=bool(data.get("dry_run", False)),
             )
             await adapters.truth_store.save_export_result(result)
             audit = engine.build_writeback_audit_event(
                 entity_id=str(entity_id),
                 result=result,
-                trigger="event:export-jobs",
+                trigger="event:export-jobs:hitl-approved",
             )
             await adapters.truth_store.save_audit_event(audit.model_dump())
             logger.info(

apps/truth-export/src/truth_export/export_engine.py

@@ -8,7 +8,7 @@
 
 from holiday_peak_lib.adapters.acp_mapper import AcpCatalogMapper
 from holiday_peak_lib.adapters.ucp_mapper import UcpProtocolMapper
-from holiday_peak_lib.integrations import PIMWritebackManager
+from holiday_peak_lib.integrations import PIMWritebackManager, WritebackResult, WritebackStatus
 
 from .schemas_compat import AuditAction, AuditEvent, ExportResult
 
@@ -118,6 +118,63 @@ async def writeback_entity(
         summary["pim_response_summary"] = self._build_pim_response_summary(summary)
         return summary
 
+    async def writeback_to_pim(
+        self,
+        manager: PIMWritebackManager,
+        truth_store: Any,
+        entity_id: str,
+        *,
+        approved_attributes: list[str] | None = None,
+        dry_run: bool = False,
+    ) -> dict[str, Any]:
+        """Write back approved attributes only, preserving manager safety guards."""
+        if not approved_attributes:
+            return await self.writeback_entity(manager, entity_id, dry_run=dry_run)
+
+        approved_set = {field for field in approved_attributes if field}
+        if dry_run:
+            preview = await self.writeback_entity(manager, entity_id, dry_run=True)
+            return self._filter_writeback_summary(preview, approved_set)
+
+        raw_attributes = await truth_store.get_attributes(entity_id)
+        selected = [
+            attr
+            for attr in raw_attributes
+            if attr.get("field") in approved_set and attr.get("writeback_eligible", False)
+        ]
+
+        if not selected:
+            skipped_summary: dict[str, Any] = {
+                "entity_id": entity_id,
+                "total": 0,
+                "succeeded": 0,
+                "skipped": 0,
+                "conflicts": 0,
+                "errors": 0,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+                "results": [],
+                "dry_run": False,
+                "status": "skipped",
+            }
+            skipped_summary["pim_response_summary"] = self._build_pim_response_summary(
+                skipped_summary
+            )
+            return skipped_summary
+
+        async def _write_one(attr: dict[str, Any]) -> WritebackResult:
+            return await manager.writeback_attribute(
+                entity_id,
+                field=str(attr.get("field")),
+                value=attr.get("value"),
+                truth_version=attr.get("version"),
+            )
+
+        results = list(await asyncio.gather(*[_write_one(attr) for attr in selected]))
+        summary = self._summarize_field_results(entity_id, results, dry_run=False)
+        summary["status"] = self._resolve_writeback_status(summary)
+        summary["pim_response_summary"] = self._build_pim_response_summary(summary)
+        return summary
+
     async def writeback_batch(
         self,
         manager: PIMWritebackManager,
@@ -185,3 +242,71 @@ def _build_pim_response_summary(result: dict[str, Any]) -> dict[str, Any]:
             "errors": result.get("errors", 0),
             "messages": messages,
         }
+
+    def _summarize_field_results(
+        self,
+        entity_id: str,
+        field_results: list[WritebackResult],
+        *,
+        dry_run: bool,
+    ) -> dict[str, Any]:
+        succeeded = 0
+        skipped = 0
+        conflicts = 0
+        errors = 0
+        result_items: list[dict[str, Any]] = []
+
+        for item in field_results:
+            status = item.status
+            if status == WritebackStatus.SUCCESS:
+                succeeded += 1
+            elif status in (WritebackStatus.SKIPPED, WritebackStatus.DRY_RUN):
+                skipped += 1
+            elif status == WritebackStatus.CONFLICT:
+                conflicts += 1
+            else:
+                errors += 1
+            result_items.append(item.model_dump())
+
+        return {
+            "entity_id": entity_id,
+            "total": len(field_results),
+            "succeeded": succeeded,
+            "skipped": skipped,
+            "conflicts": conflicts,
+            "errors": errors,
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "results": result_items,
+            "dry_run": dry_run,
+        }
+
+    def _filter_writeback_summary(
+        self,
+        summary: dict[str, Any],
+        approved_set: set[str],
+    ) -> dict[str, Any]:
+        filtered_results = [
+            result for result in summary.get("results", []) if result.get("field") in approved_set
+        ]
+        filtered = dict(summary)
+        filtered["results"] = filtered_results
+        filtered["total"] = len(filtered_results)
+        filtered["succeeded"] = len(
+            [r for r in filtered_results if r.get("status") == WritebackStatus.SUCCESS.value]
+        )
+        filtered["skipped"] = len(
+            [
+                r
+                for r in filtered_results
+                if r.get("status") in {WritebackStatus.SKIPPED.value, WritebackStatus.DRY_RUN.value}
+            ]
+        )
+        filtered["conflicts"] = len(
+            [r for r in filtered_results if r.get("status") == WritebackStatus.CONFLICT.value]
+        )
+        filtered["errors"] = len(
+            [r for r in filtered_results if r.get("status") == WritebackStatus.ERROR.value]
+        )
+        filtered["status"] = self._resolve_writeback_status(filtered)
+        filtered["pim_response_summary"] = self._build_pim_response_summary(filtered)
+        return filtered

apps/truth-export/src/truth_export/routes.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
@@ -39,12 +40,14 @@ class BulkExportRequest(BaseModel):
 class PIMWritebackRequest(BaseModel):
     dry_run: bool = False
     trigger: str = "api"
+    approved_fields: list[str] | None = None
 
 
 class PIMBatchWritebackRequest(BaseModel):
     entity_ids: list[str] = Field(min_length=1, max_length=100)
     dry_run: bool = False
     max_concurrency: int = Field(default=5, ge=1, le=20)
+    approved_fields: list[str] | None = None
 
 
 # ---------------------------------------------------------------------------
@@ -103,12 +106,29 @@ async def export_pim_batch(
     if len(request.entity_ids) > 100:
         raise HTTPException(status_code=400, detail="Batch size cannot exceed 100 entities")
 
-    results = await engine.writeback_batch(
-        adapters.writeback_manager,
-        request.entity_ids,
-        dry_run=request.dry_run,
-        max_concurrency=request.max_concurrency,
-    )
+    if request.approved_fields:
+        semaphore = asyncio.Semaphore(max(1, request.max_concurrency))
+
+        async def _run_one(entity_id: str) -> dict[str, Any]:
+            async with semaphore:
+                return await engine.writeback_to_pim(
+                    adapters.writeback_manager,
+                    adapters.truth_store,
+                    entity_id,
+                    approved_attributes=request.approved_fields,
+                    dry_run=request.dry_run,
+                )
+
+        results = list(
+            await asyncio.gather(*[_run_one(entity_id) for entity_id in request.entity_ids])
+        )
+    else:
+        results = await engine.writeback_batch(
+            adapters.writeback_manager,
+            request.entity_ids,
+            dry_run=request.dry_run,
+            max_concurrency=request.max_concurrency,
+        )
 
     for entity_result in results:
         await adapters.truth_store.save_export_result(entity_result)
@@ -140,6 +160,14 @@ async def export_pim_single(
         entity_id,
         dry_run=request.dry_run,
     )
+    if request.approved_fields:
+        result = await engine.writeback_to_pim(
+            adapters.writeback_manager,
+            adapters.truth_store,
+            entity_id,
+            approved_attributes=request.approved_fields,
+            dry_run=request.dry_run,
+        )
     await adapters.truth_store.save_export_result(result)
     audit_event = engine.build_writeback_audit_event(
         entity_id=entity_id,

apps/truth-export/tests/test_event_handlers.py

@@ -13,10 +13,13 @@
 )
 from truth_export.adapters import build_truth_export_adapters
 from truth_export.event_handlers import build_event_handlers
-from truth_export.schemas_compat import ProductStyle
+from truth_export.schemas_compat import ProductStyle, TruthAttribute
 
 
 class _StubWritebackManager:
+    def __init__(self) -> None:
+        self.called_fields: list[str] = []
+
     async def dry_run(self, entity_id: str) -> ProductWritebackResult:
         return await self.writeback_product(entity_id)
 
@@ -35,6 +38,17 @@ async def writeback_product(self, entity_id: str) -> ProductWritebackResult:
             ],
         )
 
+    async def writeback_attribute(self, entity_id: str, field: str, value, *, truth_version=None):
+        _ = value
+        _ = truth_version
+        self.called_fields.append(field)
+        return WritebackResult(
+            entity_id=entity_id,
+            field=field,
+            status=WritebackStatus.SUCCESS,
+            message="Writeback succeeded",
+        )
+
 
 @pytest.mark.asyncio
 async def test_build_event_handlers_includes_export_jobs() -> None:
@@ -45,7 +59,27 @@ async def test_build_event_handlers_includes_export_jobs() -> None:
 @pytest.mark.asyncio
 async def test_export_jobs_hitl_approval_triggers_writeback_path() -> None:
     adapters = build_truth_export_adapters()
-    adapters.writeback_manager = _StubWritebackManager()
+    stub_manager = _StubWritebackManager()
+    adapters.writeback_manager = stub_manager
+    adapters.truth_store.seed_attributes(
+        "STYLE-001",
+        [
+            TruthAttribute(
+                entityType="style",
+                entityId="STYLE-001",
+                attributeKey="title",
+                value="Approved title",
+                source="SYSTEM",
+            ),
+            TruthAttribute(
+                entityType="style",
+                entityId="STYLE-001",
+                attributeKey="description",
+                value="Should not be written",
+                source="SYSTEM",
+            ),
+        ],
+    )
 
     handlers = build_event_handlers(adapters=adapters)
     handler = handlers["export-jobs"]
@@ -59,6 +93,7 @@ async def test_export_jobs_hitl_approval_triggers_writeback_path() -> None:
                 "entity_id": "STYLE-001",
                 "protocol": "pim",
                 "status": "approved",
+                "approved_fields": ["title"],
             },
         }
     )
@@ -71,6 +106,7 @@ async def test_export_jobs_hitl_approval_triggers_writeback_path() -> None:
         adapters.truth_store._audit_events[-1]["details"]["writeback_status"]
         == "completed"  # pylint: disable=protected-access
     )
+    assert stub_manager.called_fields == ["title"]
 
 
 @pytest.mark.asyncio

apps/truth-export/tests/test_export.py

@@ -339,3 +339,59 @@ async def get_attributes(self, _entity_id: str):
     assert result["status"] == "conflict"
     assert result["conflicts"] == 1
     pim.push_enrichment.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_writeback_to_pim_uses_approved_fields_only(engine, adapters):
+    adapters.truth_store.seed_attributes(
+        "STYLE-001",
+        [
+            TruthAttribute(
+                entityType="style",
+                entityId="STYLE-001",
+                attributeKey="title",
+                value="Approved title",
+                source="SYSTEM",
+            ),
+            TruthAttribute(
+                entityType="style",
+                entityId="STYLE-001",
+                attributeKey="description",
+                value="Should not be written",
+                source="SYSTEM",
+            ),
+        ],
+    )
+
+    class _StubManager:
+        def __init__(self):
+            self.called_fields: list[str] = []
+
+        async def writeback_attribute(
+            self, entity_id: str, field: str, value, *, truth_version=None
+        ):
+            _ = entity_id
+            _ = value
+            _ = truth_version
+            self.called_fields.append(field)
+
+            from holiday_peak_lib.integrations import WritebackResult, WritebackStatus
+
+            return WritebackResult(
+                entity_id="STYLE-001",
+                field=field,
+                status=WritebackStatus.SUCCESS,
+                message="Writeback succeeded",
+            )
+
+    manager = _StubManager()
+    result = await engine.writeback_to_pim(
+        manager,
+        adapters.truth_store,
+        "STYLE-001",
+        approved_attributes=["title"],
+    )
+
+    assert manager.called_fields == ["title"]
+    assert result["status"] == "completed"
+    assert result["total"] == 1