UN-2663 [FEAT] moving away from redis hitl file handling improvements

backend/sample.env

@@ -212,5 +212,8 @@ RUNNER_POLLING_INTERVAL_SECONDS=2
 # Examples: 900 (15 min), 1800 (30 min), 3600 (60 min)
 MIN_SCHEDULE_INTERVAL_SECONDS=1800
 
+# HITL Files Storage Configuration
+HITL_FILES_FILE_STORAGE_CREDENTIALS='{"provider": "minio", "credentials": {"endpoint_url": "http://unstract-minio:9000", "key": "minio", "secret": "minio123"}}'
+
 # File active cache redis db
 FILE_ACTIVE_CACHE_REDIS_DB=0

backend/workflow_manager/endpoint_v2/destination.py

@@ -243,7 +243,6 @@ def handle_output(
         """Handle the output based on the connection type."""
         connection_type = self.endpoint.connection_type
         tool_execution_result: str | None = None
-
         if connection_type == WorkflowEndpoint.ConnectionType.FILESYSTEM:
             self.copy_output_to_output_directory()
 
@@ -893,28 +892,45 @@ def _enqueue_to_packet_or_regular_queue(
             q_name: Queue name for regular queue
             ttl_seconds: TTL in seconds (optional, for regular queue)
         """
-        # Get queue instance
-        conn = QueueUtils.get_queue_inst()
+        # Get organization_id for proper HITL queue filtering
+        organization_id = UserContext.get_organization_identifier()
 
+        # Get HITL backend queue instance with organization context
+        conn = QueueUtils.get_queue_inst(
+            {"use_hitl_backend": True, "organization_id": organization_id}
+        )
+
+        # Check if this is packet processing
         if self.packet_id:
-            # Route to packet queue
+            # Route to packet queue (Enterprise feature)
+            logger.info(f"Routing {file_name} to packet queue {self.packet_id}")
             success = conn.enqueue_to_packet(
-                packet_id=self.packet_id, queue_result=queue_result
+                packet_id=self.packet_id,
+                queue_result=queue_result,
+                organization_id=organization_id,
             )
             if not success:
                 error_msg = f"Failed to push {file_name} to packet {self.packet_id}"
                 logger.error(error_msg)
                 raise RuntimeError(error_msg)
+            logger.info(
+                f"Pushed {file_name} to packet {self.packet_id} with organization_id {organization_id}"
+            )
             return
 
         # Route to regular queue
         if ttl_seconds:
             conn.enqueue_with_ttl(
                 queue_name=q_name, message=queue_result_json, ttl_seconds=ttl_seconds
             )
+            logger.info(
+                f"Pushed {file_name} to queue {q_name} with organization_id {organization_id} and TTL {ttl_seconds}"
+            )
         else:
             conn.enqueue(queue_name=q_name, message=queue_result_json)
-        logger.info(f"Pushed {file_name} to queue {q_name} with file content")
+            logger.info(
+                f"Pushed {file_name} to queue {q_name} with organization_id {organization_id}"
+            )
 
     def _create_queue_result(
         self,

backend/workflow_manager/endpoint_v2/queue_utils.py

@@ -4,6 +4,7 @@
 from enum import Enum
 from typing import Any
 
+from django.conf import settings
 from utils.constants import Common
 from workflow_manager.endpoint_v2.exceptions import UnstractQueueException
 
@@ -20,8 +21,25 @@ class QueueResultStatus(Enum):
 
 
 class QueueUtils:
+    # HITL connectors registry (lazy-loaded to avoid circular imports)
+    _hitl_connectors = {}
+
     @staticmethod
     def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:
+        """Get queue connector instance based on configuration.
+
+        For HITL operations, this can return PostgreSQL, Hybrid, or Redis connectors
+        based on the HITL_QUEUE_BACKEND setting.
+        """
+        # Check if this is for HITL operations
+        is_hitl = connector_settings.get("use_hitl_backend", False)
+
+        if is_hitl:
+            # Use HITL-specific queue backend
+            hitl_backend = getattr(settings, "HITL_QUEUE_BACKEND", "hybrid")
+            return QueueUtils.get_hitl_queue_inst(hitl_backend, connector_settings)
+
+        # Default behavior for non-HITL operations
         if not queue_connectors:
             raise UnstractQueueException(detail="Queue connector not exists")
         queue_connector_key = next(iter(queue_connectors))
@@ -31,6 +49,108 @@ def get_queue_inst(connector_settings: dict[str, Any] = {}) -> UnstractQueue:
         connector_class: UnstractQueue = connector(connector_settings)
         return connector_class
 
+    @staticmethod
+    def get_hitl_queue_inst(
+        backend: str, connector_settings: dict[str, Any] = {}
+    ) -> UnstractQueue:
+        """Get HITL-specific queue connector instance with dynamic imports.
+
+        This method uses dynamic imports to avoid hard dependencies on the
+        manual_review_v2 pluggable app, allowing graceful degradation when
+        the app is not available.
+
+        Args:
+            backend: Backend type ('postgresql', 'hybrid', 'redis')
+            connector_settings: Connector configuration
+
+        Returns:
+            Configured queue connector instance
+
+        Raises:
+            UnstractQueueException: When HITL connectors are not available
+        """
+        # For Redis backend, use default connector
+        if backend == "redis":
+            # Strip HITL flag to force default (non-HITL) connector path
+            non_hitl_settings = dict(connector_settings)
+            non_hitl_settings.pop("use_hitl_backend", None)
+            return QueueUtils.get_queue_inst(non_hitl_settings)
+
+        # For PostgreSQL and Hybrid backends, try dynamic imports
+        try:
+            if backend == "postgresql":
+                connector_class = QueueUtils._import_hitl_connector("PostgreSQLQueue")
+                return connector_class(connector_settings)
+
+            elif backend == "hybrid":
+                connector_class = QueueUtils._import_hitl_connector("HybridQueue")
+                return connector_class(connector_settings)
+
+            else:
+                logger.warning(
+                    f"Unknown HITL queue backend '{backend}'. "
+                    f"Valid options: postgresql, hybrid, redis. "
+                    f"Attempting fallback to hybrid."
+                )
+                connector_class = QueueUtils._import_hitl_connector("HybridQueue")
+                return connector_class(connector_settings)
+
+        except ImportError as e:
+            logger.error(
+                f"HITL queue backend '{backend}' not available: {e}. "
+                f"Make sure 'pluggable_apps.manual_review_v2' is installed and configured."
+            )
+            raise UnstractQueueException(
+                detail=f"HITL queue backend '{backend}' not available. "
+                f"Please install the manual_review_v2 app or use 'redis' backend."
+            )
+        except Exception as e:
+            logger.error(f"Failed to initialize HITL queue backend '{backend}': {e}")
+            raise UnstractQueueException(
+                detail=f"Failed to initialize HITL queue backend '{backend}': {str(e)}"
+            )
+
+    @classmethod
+    def _import_hitl_connector(cls, connector_name: str):
+        """Get HITL connector class from registry (lazy-loaded to avoid circular imports).
+
+        Args:
+            connector_name: Name of the connector class to get
+
+        Returns:
+            The connector class
+
+        Raises:
+            ImportError: When the connector is not available
+        """
+        # Lazy initialization on first access (avoids circular imports)
+        if not cls._hitl_connectors:
+            try:
+                from pluggable_apps.manual_review_v2.connectors import (
+                    HybridQueue,
+                    PostgreSQLQueue,
+                )
+
+                cls._hitl_connectors = {
+                    "PostgreSQLQueue": PostgreSQLQueue,
+                    "HybridQueue": HybridQueue,
+                }
+            except ImportError as e:
+                logger.debug(f"HITL connectors not available: {e}")
+
+        if not cls._hitl_connectors:
+            raise ImportError(
+                "HITL connectors not available. Make sure 'pluggable_apps.manual_review_v2' is installed."
+            )
+
+        if connector_name not in cls._hitl_connectors:
+            available_connectors = list(cls._hitl_connectors.keys())
+            raise ImportError(
+                f"Unknown HITL connector: {connector_name}. Available: {available_connectors}"
+            )
+
+        return cls._hitl_connectors[connector_name]
+
     @staticmethod
     def calculate_remaining_ttl(enqueued_at: float, ttl_seconds: int) -> int | None:
         """Calculate remaining TTL based on original enqueue time and TTL.

unstract/filesystem/src/unstract/filesystem/file_storage_config.py

@@ -34,9 +34,11 @@ def get_provider(var_name: str, default: str = "minio") -> FileStorageProvider:
 STORAGE_MAPPING = {
     FileStorageType.WORKFLOW_EXECUTION: StorageType.SHARED_TEMPORARY,
     FileStorageType.API_EXECUTION: StorageType.SHARED_TEMPORARY,
+    FileStorageType.HITL_FILES: StorageType.SHARED_TEMPORARY,  # Use shared temporary for HITL files
 }
 
 FILE_STORAGE_CREDENTIALS_TO_ENV_NAME_MAPPING = {
     FileStorageType.WORKFLOW_EXECUTION: "WORKFLOW_EXECUTION_FILE_STORAGE_CREDENTIALS",
     FileStorageType.API_EXECUTION: "API_FILE_STORAGE_CREDENTIALS",
+    FileStorageType.HITL_FILES: "HITL_FILES_FILE_STORAGE_CREDENTIALS",
 }

unstract/filesystem/src/unstract/filesystem/file_storage_types.py

@@ -4,3 +4,4 @@
 class FileStorageType(Enum):
     WORKFLOW_EXECUTION = "WORKFLOW_EXECUTION"
     API_EXECUTION = "API_EXECUTION"
+    HITL_FILES = "HITL_FILES"