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)
+        packet_id = serializer.validated_data.get(ApiExecution.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,
+            packet_id=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"
+    PACKET_ID: str = "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,
+        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
+            packet_id (str, optional): Packet ID for packet-based review
 
         Returns:
             ReturnDict: execution status/ result
@@ -178,6 +180,10 @@ def execute_workflow(
             logger.info(
                 f"API execution with HITL: hitl_queue_name={hitl_queue_name}, api_name={api.api_name}"
             )
+        if packet_id:
+            logger.info(
+                f"API execution with Packet: packet_id={packet_id}, api_name={api.api_name}"
+            )
         tags = Tag.bulk_get_or_create(tag_names=tag_names)
         workflow_execution = WorkflowExecutionServiceHelper.create_workflow_execution(
             workflow_id=workflow_id,
@@ -234,6 +240,7 @@ def execute_workflow(
                 use_file_history=use_file_history,
                 llm_profile_id=llm_profile_id,
                 hitl_queue_name=hitl_queue_name,
+                packet_id=packet_id,
             )
             result.status_api = DeploymentHelper.construct_status_endpoint(
                 api_endpoint=api.api_endpoint, execution_id=execution_id

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,19 @@ def _should_handle_hitl(
             logger.info(f"Successfully pushed {file_name} to HITL queue")
             return True
 
+        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,
+            )
+            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 +765,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 +794,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
@@ -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 success:
+                    logger.info(f"Pushed {file_name} to packet {self.packet_id}")
+                else:
+                    logger.error(f"Failed to push {file_name} to packet {self.packet_id}")
+                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 success:
+                    logger.info(f"Pushed {file_name} to packet {self.packet_id}")
+                else:
+                    logger.error(f"Failed to push {file_name} to packet {self.packet_id}")
+                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