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

backend/api_v2/api_deployment_views.py

@@ -70,6 +70,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)
 
         if presigned_urls:
             DeploymentHelper.load_presigned_files(presigned_urls, file_objs)
@@ -85,6 +86,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,
             request_headers=dict(request.headers),
         )
         if "error" in response and response["error"]:

backend/api_v2/constants.py

@@ -10,4 +10,5 @@ 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"

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,
         request_headers=None,
     ) -> ReturnDict:
         """Execute workflow by api.
@@ -168,6 +169,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
 
         Returns:
             ReturnDict: execution status/ result
@@ -234,6 +236,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,
             )
             result.status_api = DeploymentHelper.construct_status_endpoint(
                 api_endpoint=api.api_endpoint, execution_id=execution_id

backend/api_v2/serializers.py

@@ -224,6 +224,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)
 
     def validate_hitl_queue_name(self, value: str | None) -> str | None:
         """Validate queue name format using enterprise validation if available."""

backend/workflow_manager/endpoint_v2/destination.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from connector_v2.models import ConnectorInstance
+from pluggable_apps.manual_review_v2.packet_queue_utils import PacketQueueUtils
 from plugins.workflow_manager.workflow_v2.utils import WorkflowUtil
 from rest_framework.exceptions import APIException
 from usage_v2.helper import UsageHelper
@@ -64,6 +65,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.
 
@@ -82,6 +84,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(
@@ -172,6 +175,16 @@ def _should_handle_hitl(
             logger.info(f"Successfully pushed {file_name} to HITL queue")
             return True
 
+        if self.packet_id:
+            self._push_data_to_queue(
+                file_name=file_name,
+                workflow=workflow,
+                input_file_path=input_file_path,
+                file_execution_id=file_execution_id,
+            )
+            logger.info(f"Successfully pushed {file_name} to packet queue")
+            return True
+
         # Skip HITL validation if we're using file_history and no execution result is available
         if self.is_api and self.use_file_history:
             return False
@@ -749,6 +762,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
@@ -777,6 +791,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
@@ -836,7 +851,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:
@@ -868,6 +886,18 @@ def _push_to_queue(
 
             queue_result_json = json.dumps(queue_result)
 
+            # Check if this is a packet-based execution
+            if self.packet_id:
+                # Route to packet queue instead of regular HITL queue
+                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")
@@ -918,6 +948,18 @@ def _push_to_queue(
                 )
                 raise ValueError("Cannot enqueue empty JSON message")
 
+            # Check if this is a packet-based execution
+            if self.packet_id:
+                # Route to packet queue instead of regular HITL queue
+                success = PacketQueueUtils.enqueue_to_packet(
+                    packet_id=self.packet_id, queue_result=queue_result_obj
+                )
+                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",
 }
 
 
@@ -267,6 +268,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,
     ) -> ExecutionResponse:
         tool_instances: list[ToolInstance] = (
             ToolInstanceHelper.get_tool_instances_by_workflow(
@@ -295,6 +297,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
@@ -436,6 +439,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,
     ) -> ExecutionResponse:
         """Adding a workflow to the queue for execution.
 
@@ -449,6 +453,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
@@ -475,6 +480,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,
                 },
                 queue=queue,
             )
@@ -678,8 +684,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,
@@ -692,6 +699,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,
             )
         except Exception as error:
             error_message = traceback.format_exc()