UN-2807 [FEAT] Add packet processing support for HITL workflows

backend/api_v2/api_deployment_views.py

@@ -81,6 +81,7 @@ def post(
         tag_names = serializer.validated_data.get(ApiExecution.TAGS)
         llm_profile_id = serializer.validated_data.get(ApiExecution.LLM_PROFILE_ID)
         hitl_queue_name = serializer.validated_data.get(ApiExecution.HITL_QUEUE_NAME)
+        hitl_packet_id = serializer.validated_data.get(ApiExecution.HITL_PACKET_ID)
         custom_data = serializer.validated_data.get(ApiExecution.CUSTOM_DATA)
 
         if presigned_urls:
@@ -97,6 +98,7 @@ def post(
             tag_names=tag_names,
             llm_profile_id=llm_profile_id,
             hitl_queue_name=hitl_queue_name,
+            hitl_packet_id=hitl_packet_id,
             custom_data=custom_data,
             request_headers=dict(request.headers),
         )

backend/api_v2/constants.py

@@ -10,5 +10,6 @@ class ApiExecution:
     TAGS: str = "tags"
     LLM_PROFILE_ID: str = "llm_profile_id"
     HITL_QUEUE_NAME: str = "hitl_queue_name"
+    HITL_PACKET_ID: str = "hitl_packet_id"
     PRESIGNED_URLS: str = "presigned_urls"
     CUSTOM_DATA: str = "custom_data"

backend/api_v2/deployment_helper.py

@@ -155,6 +155,7 @@ def execute_workflow(
         tag_names: list[str] = [],
         llm_profile_id: str | None = None,
         hitl_queue_name: str | None = None,
+        hitl_packet_id: str | None = None,
         custom_data: dict[str, Any] | None = None,
         request_headers=None,
     ) -> ReturnDict:
@@ -169,6 +170,7 @@ def execute_workflow(
             tag_names (list(str)): list of tag names
             llm_profile_id (str, optional): LLM profile ID for overriding tool settings
             hitl_queue_name (str, optional): Custom queue name for manual review
+            hitl_packet_id (str, optional): Packet ID for packet-based review
             custom_data (dict[str, Any], optional): JSON data for custom_data variable replacement in prompts
 
         Returns:
@@ -236,6 +238,7 @@ def execute_workflow(
                 use_file_history=use_file_history,
                 llm_profile_id=llm_profile_id,
                 hitl_queue_name=hitl_queue_name,
+                hitl_packet_id=hitl_packet_id,
                 custom_data=custom_data,
             )
             result.status_api = DeploymentHelper.construct_status_endpoint(

backend/api_v2/serializers.py

@@ -6,6 +6,7 @@
 
 from django.core.validators import RegexValidator
 from pipeline_v2.models import Pipeline
+from pluggable_apps.feature_registry import FeatureRegistry
 from prompt_studio.prompt_profile_manager_v2.models import ProfileManager
 from rest_framework.serializers import (
     BooleanField,
@@ -227,6 +228,7 @@ class ExecutionRequestSerializer(TagParamsSerializer):
     presigned_urls = ListField(child=URLField(), required=False)
     llm_profile_id = CharField(required=False, allow_null=True, allow_blank=True)
     hitl_queue_name = CharField(required=False, allow_null=True, allow_blank=True)
+    hitl_packet_id = CharField(required=False, allow_null=True, allow_blank=True)
     custom_data = JSONField(required=False, allow_null=True)
 
     def validate_hitl_queue_name(self, value: str | None) -> str | None:
@@ -248,6 +250,20 @@ def validate_hitl_queue_name(self, value: str | None) -> str | None:
             )
         return value
 
+    def validate_hitl_packet_id(self, value: str | None) -> str | None:
+        """Validate packet ID format using enterprise validation if available."""
+        if not value:
+            return value
+
+        # Check if HITL feature is available using FeatureRegistry
+        if not FeatureRegistry.is_hitl_available():
+            raise ValidationError(
+                "Packet-based HITL processing requires Unstract Enterprise. "
+                "This advanced workflow feature is available in our enterprise version. "
+                "Learn more at https://docs.unstract.com/unstract/unstract_platform/features/workflows/hqr_deployment_workflows/ or "
+                "contact our sales team at https://unstract.com/contact/"
+            )
+
     def validate_custom_data(self, value):
         """Validate custom_data is a valid JSON object."""
         if value is None:

backend/workflow_manager/endpoint_v2/destination.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from connector_v2.models import ConnectorInstance
+from pluggable_apps.feature_registry import FeatureRegistry
 from plugins.workflow_manager.workflow_v2.utils import WorkflowUtil
 from rest_framework.exceptions import APIException
 from usage_v2.helper import UsageHelper
@@ -70,6 +71,7 @@ def __init__(
         use_file_history: bool,
         file_execution_id: str | None = None,
         hitl_queue_name: str | None = None,
+        packet_id: str | None = None,
     ) -> None:
         """Initialize a DestinationConnector object.
 
@@ -88,6 +90,7 @@ def __init__(
         self.workflow_log = workflow_log
         self.use_file_history = use_file_history
         self.hitl_queue_name = hitl_queue_name
+        self.packet_id = packet_id
         self.workflow = workflow
 
     def _get_endpoint_for_workflow(
@@ -166,6 +169,19 @@ def _should_handle_hitl(
         file_execution_id: str,
     ) -> bool:
         """Determines if HITL processing should be performed, returning True if data was pushed to the queue."""
+        # Check packet_id first - it takes precedence over hitl_queue_name
+        if self.packet_id:
+            logger.info(
+                f"API packet override: pushing to packet queue for file {file_name}"
+            )
+            self._push_data_to_queue(
+                file_name=file_name,
+                workflow=workflow,
+                input_file_path=input_file_path,
+                file_execution_id=file_execution_id,
+            )
+            return True
+
         # Check if API deployment requested HITL override
         if self.hitl_queue_name:
             logger.info(f"API HITL override: pushing to queue for file {file_name}")
@@ -175,7 +191,6 @@ def _should_handle_hitl(
                 input_file_path=input_file_path,
                 file_execution_id=file_execution_id,
             )
-            logger.info(f"Successfully pushed {file_name} to HITL queue")
             return True
 
         # Skip HITL validation if we're using file_history and no execution result is available
@@ -753,6 +768,7 @@ def get_config(self) -> DestinationConfig:
             execution_id=self.execution_id,
             use_file_history=self.use_file_history,
             hitl_queue_name=self.hitl_queue_name,
+            packet_id=self.packet_id,
         )
 
     @classmethod
@@ -781,6 +797,7 @@ def from_config(
             use_file_history=config.use_file_history,
             file_execution_id=config.file_execution_id,
             hitl_queue_name=config.hitl_queue_name,
+            packet_id=config.packet_id,
         )
 
         return destination
@@ -840,7 +857,10 @@ def _push_to_queue(
             None
         """
         if not result:
-            return
+            if not self.packet_id:
+                return
+            # For packet processing, use a placeholder result if none available
+            result = json.dumps({"status": "pending", "message": "Awaiting processing"})
         connector: ConnectorInstance = self.source_endpoint.connector_instance
         # For API deployments, use workflow execution storage instead of connector
         if self.is_api:
@@ -871,6 +891,28 @@ def _push_to_queue(
             ).to_dict()
 
             queue_result_json = json.dumps(queue_result)
+
+            # Check if this is a packet-based execution
+            if self.packet_id:
+                if not FeatureRegistry.is_hitl_available():
+                    raise ValueError(
+                        "Packet-based HITL processing requires Unstract Enterprise. "
+                        "This feature is not available in the OSS version."
+                    )
+                # Route to packet queue instead of regular HITL queue
+                from pluggable_apps.manual_review_v2.packet_queue_utils import (
+                    PacketQueueUtils,
+                )
+
+                success = PacketQueueUtils.enqueue_to_packet(
+                    packet_id=self.packet_id, queue_result=queue_result
+                )
+                if not success:
+                    error_msg = f"Failed to push {file_name} to packet {self.packet_id}"
+                    logger.error(error_msg)
+                    raise RuntimeError(error_msg)
+                return
+
             conn = QueueUtils.get_queue_inst()
             conn.enqueue(queue_name=q_name, message=queue_result_json)
             logger.info(f"Pushed {file_name} to queue {q_name} with file content")
@@ -923,6 +965,27 @@ def _push_to_queue(
                 )
                 raise ValueError("Cannot enqueue empty JSON message")
 
+            # Check if this is a packet-based execution
+            if self.packet_id:
+                if not FeatureRegistry.is_hitl_available():
+                    raise ValueError(
+                        "Packet-based HITL processing requires Unstract Enterprise. "
+                        "This feature is not available in the OSS version."
+                    )
+                # Route to packet queue instead of regular HITL queue
+                from pluggable_apps.manual_review_v2.packet_queue_utils import (
+                    PacketQueueUtils,
+                )
+
+                success = PacketQueueUtils.enqueue_to_packet(
+                    packet_id=self.packet_id, queue_result=queue_result
+                )
+                if not success:
+                    error_msg = f"Failed to push {file_name} to packet {self.packet_id}"
+                    logger.error(error_msg)
+                    raise RuntimeError(error_msg)
+                return
+
             conn = QueueUtils.get_queue_inst()
 
             # Use the TTL metadata that was already set in the QueueResult object

backend/workflow_manager/endpoint_v2/dto.py

@@ -92,6 +92,7 @@ class DestinationConfig:
     use_file_history: bool
     file_execution_id: str | None = None
     hitl_queue_name: str | None = None
+    packet_id: str | None = None
 
     def to_json(self) -> dict[str, Any]:
         """Serialize the DestinationConfig instance to a JSON string."""
@@ -104,6 +105,7 @@ def to_json(self) -> dict[str, Any]:
             "use_file_history": self.use_file_history,
             "file_execution_id": file_execution_id,
             "hitl_queue_name": self.hitl_queue_name,
+            "packet_id": self.packet_id,
         }
 
     @staticmethod

backend/workflow_manager/workflow_v2/workflow_helper.py

@@ -64,6 +64,7 @@
 EXECUTION_EXCLUDED_PARAMS = {
     "llm_profile_id",
     "hitl_queue_name",
+    "hitl_packet_id",
     "custom_data",
 }
 
@@ -268,6 +269,7 @@ def run_workflow(
         use_file_history: bool = True,
         llm_profile_id: str | None = None,
         hitl_queue_name: str | None = None,
+        packet_id: str | None = None,
         custom_data: dict[str, Any] | None = None,
     ) -> ExecutionResponse:
         tool_instances: list[ToolInstance] = (
@@ -297,6 +299,7 @@ def run_workflow(
             workflow_log=workflow_log,
             use_file_history=use_file_history,
             hitl_queue_name=hitl_queue_name,
+            packet_id=packet_id,
         )
         try:
             # Validating endpoints
@@ -439,6 +442,7 @@ def execute_workflow_async(
         use_file_history: bool = True,
         llm_profile_id: str | None = None,
         hitl_queue_name: str | None = None,
+        hitl_packet_id: str | None = None,
         custom_data: dict[str, Any] | None = None,
     ) -> ExecutionResponse:
         """Adding a workflow to the queue for execution.
@@ -453,6 +457,7 @@ def execute_workflow_async(
                 processed files. Defaults to True
             hitl_queue_name (str | None): Name of the HITL queue to push files to
             llm_profile_id (str, optional): LLM profile ID for overriding tool settings
+            hitl_packet_id (str | None): Packet ID for packet-based HITL workflows
 
         Returns:
             ExecutionResponse: Existing status of execution
@@ -479,6 +484,7 @@ def execute_workflow_async(
                     "use_file_history": use_file_history,
                     "llm_profile_id": llm_profile_id,
                     "hitl_queue_name": hitl_queue_name,
+                    "hitl_packet_id": hitl_packet_id,
                     "custom_data": custom_data,
                 },
                 queue=queue,
@@ -679,8 +685,9 @@ def execute_workflow(
             execution_id=execution_id, task_id=task_id
         )
         try:
+            hitl_packet_id_from_kwargs = kwargs.get("hitl_packet_id")
             logger.info(
-                f"Starting workflow execution: workflow_id={workflow_id}, execution_id={execution_id}, hitl_queue_name={kwargs.get('hitl_queue_name')}"
+                f"Starting workflow execution: workflow_id={workflow_id}, execution_id={execution_id}, hitl_queue_name={kwargs.get('hitl_queue_name')}, hitl_packet_id={hitl_packet_id_from_kwargs}"
             )
             execution_response = WorkflowHelper.run_workflow(
                 workflow=workflow,
@@ -693,6 +700,7 @@ def execute_workflow(
                 use_file_history=use_file_history,
                 llm_profile_id=kwargs.get("llm_profile_id"),
                 hitl_queue_name=kwargs.get("hitl_queue_name"),
+                packet_id=hitl_packet_id_from_kwargs,
                 custom_data=kwargs.get("custom_data"),
             )
         except Exception as error: