feat: Truth HITL service — human-in-the-loop review queue for AI-proposed attributes (#126)

* Initial plan

* feat(#102): implement Truth HITL service

Co-authored-by: Cataldir <29005497+Cataldir@users.noreply.github.com>

* chore: merge main into branch, resolve conflicts

Co-authored-by: Cataldir <29005497+Cataldir@users.noreply.github.com>

* fix: align truth-hitl imports with lint policy

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: Cataldir <29005497+Cataldir@users.noreply.github.com>
Co-authored-by: Ricardo Cataldi <rcataldi@microsoft.com>

Author: Copilot

apps/truth-hitl/src/Dockerfile

@@ -0,0 +1,29 @@
+# Generated for truth-hitl
+ARG PYTHON_VERSION=3.13-slim
+
+FROM python:${PYTHON_VERSION} AS base
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1 \
+    APP_NAME=truth-hitl
+WORKDIR /app/src
+RUN apt-get update && apt-get install -y --no-install-recommends build-essential libpq-dev git curl \
+    && rm -rf /var/lib/apt/lists/*
+
+FROM base AS dev
+ENV APP_ENV=dev
+COPY pyproject.toml ./pyproject.toml
+COPY . ./
+ENV PYTHONPATH=/app/src
+RUN python -m pip install --upgrade pip && \
+    python -m pip install --no-cache-dir -e .[dev,test,lint]
+CMD ["uvicorn", "truth_hitl.main:app", "--host", "0.0.0.0", "--port", "8000", "--reload"]
+
+FROM base AS prod
+ENV APP_ENV=prod
+COPY pyproject.toml ./pyproject.toml
+COPY . ./
+ENV PYTHONPATH=/app/src
+RUN python -m pip install --upgrade pip && \
+    python -m pip install --no-cache-dir .
+CMD ["uvicorn", "truth_hitl.main:app", "--host", "0.0.0.0", "--port", "8000"]

apps/truth-hitl/src/pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "truth-hitl"
+version = "0.1.0"
+description = "Truth Human-in-the-Loop review service"
+authors = [{name = "Ricardo Cataldi", email = "rcataldi@microsoft.com"}]
+requires-python = ">=3.13"
+dependencies = ["fastapi", "fastapi-mcp", "uvicorn", "pydantic>=2", "agent-framework", "azure-ai-projects>=2.0.0b4", "azure-identity", "azure-search-documents", "azure-cosmos", "azure-storage-blob", "azure-eventhub", "redis", "asyncpg", "holiday-peak-lib @ git+https://github.com/Azure-Samples/holiday-peak-hub.git@v1.0.0#subdirectory=lib/src"]
+
+[project.optional-dependencies]
+dev = ["faker", "pre-commit", "python-dotenv", "debugpy"]
+test = ["pytest", "pytest-cov", "pytest-asyncio", "httpx", "requests"]
+lint = ["pylint", "isort", "black[jupyter]>=25.1.0"]
+
+[tool.pytest.ini_options]
+addopts = "-ra -q -s --cov=truth_hitl --cov-report=term-missing"
+testpaths = ["tests"]
+
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"

apps/truth-hitl/src/truth_hitl/__init__.py

@@ -0,0 +1 @@
+"""Truth HITL service package."""

apps/truth-hitl/src/truth_hitl/adapters.py

@@ -0,0 +1,24 @@
+"""Adapters for the Truth HITL service."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from truth_hitl.review_manager import ReviewManager
+
+
+@dataclass
+class HITLAdapters:
+    """Container for Truth HITL service adapters."""
+
+    review_manager: ReviewManager = field(default_factory=ReviewManager)
+
+
+def build_hitl_adapters(
+    *,
+    review_manager: ReviewManager | None = None,
+) -> HITLAdapters:
+    """Create adapters for the HITL review workflow."""
+    return HITLAdapters(
+        review_manager=review_manager or ReviewManager(),
+    )

apps/truth-hitl/src/truth_hitl/agents.py

@@ -0,0 +1,59 @@
+"""Truth HITL agent implementation and MCP tool registration."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from holiday_peak_lib.agents import BaseRetailAgent
+from holiday_peak_lib.agents.fastapi_mcp import FastAPIMCPServer
+
+from .adapters import HITLAdapters, build_hitl_adapters
+
+
+class TruthHITLAgent(BaseRetailAgent):
+    """Agent that manages the human-in-the-loop review queue for AI-proposed attributes."""
+
+    def __init__(self, config: Any, *args: Any, **kwargs: Any) -> None:
+        super().__init__(config, *args, **kwargs)
+        self._adapters = build_hitl_adapters()
+
+    @property
+    def adapters(self) -> HITLAdapters:
+        return self._adapters
+
+    async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
+        action = request.get("action", "stats")
+
+        if action == "stats":
+            return {"stats": self._adapters.review_manager.stats()}
+
+        entity_id = request.get("entity_id")
+        if action == "list" and entity_id:
+            items = self._adapters.review_manager.get_by_entity(entity_id)
+            return {"entity_id": entity_id, "items": [i.model_dump() for i in items]}
+
+        return {"error": "unsupported action or missing entity_id", "action": action}
+
+
+def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
+    """Expose MCP tools for the HITL review workflow."""
+    adapters = getattr(agent, "adapters", build_hitl_adapters())
+
+    async def get_review_queue(payload: dict[str, Any]) -> dict[str, Any]:
+        entity_id = payload.get("entity_id")
+        skip = int(payload.get("skip", 0))
+        limit = int(payload.get("limit", 50))
+        items = adapters.review_manager.list_pending(entity_id=entity_id, skip=skip, limit=limit)
+        return {"items": [i.model_dump() for i in items], "count": len(items)}
+
+    async def get_review_stats(payload: dict[str, Any]) -> dict[str, Any]:  # noqa: ARG001
+        return {"stats": adapters.review_manager.stats()}
+
+    async def get_audit_log(payload: dict[str, Any]) -> dict[str, Any]:
+        entity_id = payload.get("entity_id")
+        events = adapters.review_manager.audit_log(entity_id=entity_id)
+        return {"events": [e.model_dump() for e in events], "count": len(events)}
+
+    mcp.add_tool("/hitl/queue", get_review_queue)
+    mcp.add_tool("/hitl/stats", get_review_stats)
+    mcp.add_tool("/hitl/audit", get_audit_log)

apps/truth-hitl/src/truth_hitl/event_handlers.py

@@ -0,0 +1,67 @@
+"""Event handlers for the Truth HITL service."""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+
+from holiday_peak_lib.utils.event_hub import EventHandler
+from holiday_peak_lib.utils.logging import configure_logging
+
+from .adapters import build_hitl_adapters
+from .review_manager import ReviewItem
+
+
+def build_event_handlers() -> dict[str, EventHandler]:
+    """Build event handlers for hitl-jobs Event Hub subscription."""
+    logger = configure_logging(app_name="truth-hitl-events")
+    adapters = build_hitl_adapters()
+
+    async def handle_hitl_job(_partition_context, event) -> None:  # noqa: ANN001
+        payload = json.loads(event.body_as_str())
+        data = payload.get("data", {}) if isinstance(payload, dict) else {}
+
+        entity_id = data.get("entity_id")
+        attr_id = data.get("attr_id")
+
+        if not entity_id or not attr_id:
+            logger.info(
+                "hitl_event_skipped event_type=%s reason=missing_entity_id_or_attr_id",
+                payload.get("event_type"),
+            )
+            return
+
+        proposed_at_raw = data.get("proposed_at")
+        try:
+            proposed_at = (
+                datetime.fromisoformat(proposed_at_raw)
+                if proposed_at_raw
+                else datetime.now(timezone.utc)
+            )
+        except ValueError:
+            proposed_at = datetime.now(timezone.utc)
+
+        item = ReviewItem(
+            entity_id=entity_id,
+            attr_id=attr_id,
+            field_name=data.get("field_name", ""),
+            proposed_value=data.get("proposed_value"),
+            confidence=float(data.get("confidence", 0.0)),
+            current_value=data.get("current_value"),
+            source=data.get("source", "ai"),
+            proposed_at=proposed_at,
+            product_title=data.get("product_title", ""),
+            category_label=data.get("category_label", ""),
+            status="pending_review",
+        )
+
+        adapters.review_manager.enqueue(item)
+        logger.info(
+            "hitl_event_enqueued entity_id=%s attr_id=%s field_name=%s confidence=%.2f",
+            entity_id,
+            attr_id,
+            item.field_name,
+            item.confidence,
+        )
+
+    return {"hitl-jobs": handle_hitl_job}

apps/truth-hitl/src/truth_hitl/main.py

@@ -0,0 +1,84 @@
+"""Truth HITL service entrypoint."""
+
+import os
+
+from holiday_peak_lib.agents import FoundryAgentConfig
+from holiday_peak_lib.agents.memory import ColdMemory, HotMemory, WarmMemory
+from holiday_peak_lib.app_factory import build_service_app
+from holiday_peak_lib.config import MemorySettings
+from holiday_peak_lib.utils import EventHubSubscription, create_eventhub_lifespan
+from truth_hitl.adapters import build_hitl_adapters
+from truth_hitl.agents import TruthHITLAgent, register_mcp_tools
+from truth_hitl.event_handlers import build_event_handlers
+from truth_hitl.routes import build_review_router
+
+SERVICE_NAME = "truth-hitl"
+memory_settings = MemorySettings()
+endpoint = os.getenv("PROJECT_ENDPOINT") or os.getenv("FOUNDRY_ENDPOINT")
+project_name = os.getenv("PROJECT_NAME") or os.getenv("FOUNDRY_PROJECT_NAME")
+stream = (os.getenv("FOUNDRY_STREAM") or "").lower() in {"1", "true", "yes"}
+slm_agent_id = os.getenv("FOUNDRY_AGENT_ID_FAST")
+llm_agent_id = os.getenv("FOUNDRY_AGENT_ID_RICH")
+slm_deployment = os.getenv("MODEL_DEPLOYMENT_NAME_FAST")
+llm_deployment = os.getenv("MODEL_DEPLOYMENT_NAME_RICH")
+
+slm_config = (
+    FoundryAgentConfig(
+        endpoint=endpoint,
+        agent_id=slm_agent_id,
+        deployment_name=slm_deployment,
+        project_name=project_name,
+        stream=stream,
+    )
+    if endpoint and slm_agent_id
+    else None
+)
+
+llm_config = (
+    FoundryAgentConfig(
+        endpoint=endpoint,
+        agent_id=llm_agent_id,
+        deployment_name=llm_deployment,
+        project_name=project_name,
+        stream=stream,
+    )
+    if endpoint and llm_agent_id
+    else None
+)
+
+app = build_service_app(
+    SERVICE_NAME,
+    agent_class=TruthHITLAgent,
+    hot_memory=(HotMemory(memory_settings.redis_url) if memory_settings.redis_url else None),
+    warm_memory=(
+        WarmMemory(
+            memory_settings.cosmos_account_uri,
+            memory_settings.cosmos_database,
+            memory_settings.cosmos_container,
+        )
+        if memory_settings.cosmos_account_uri
+        else None
+    ),
+    cold_memory=(
+        ColdMemory(
+            memory_settings.blob_account_url,
+            memory_settings.blob_container,
+        )
+        if memory_settings.blob_account_url
+        else None
+    ),
+    slm_config=slm_config,
+    llm_config=llm_config,
+    mcp_setup=register_mcp_tools,
+    lifespan=create_eventhub_lifespan(
+        service_name=SERVICE_NAME,
+        subscriptions=[
+            EventHubSubscription("hitl-jobs", "hitl-service"),
+        ],
+        handlers=build_event_handlers(),
+    ),
+)
+
+# Mount the review REST routes
+_adapters = build_hitl_adapters()
+app.include_router(build_review_router(_adapters))