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

backend/sample.env

@@ -200,3 +200,5 @@ RUNNER_POLLING_INTERVAL_SECONDS=2
 # Default: 1800 seconds (30 minutes)
 # Examples: 900 (15 min), 1800 (30 min), 3600 (60 min)
 MIN_SCHEDULE_INTERVAL_SECONDS=1800
+
+HITL_FILES_FILE_STORAGE_CREDENTIALS='{"provider": "minio", "credentials": {"endpoint_url": "http://unstract-minio:9000", "key": "minio", "secret": "minio123"}}'

backend/workflow_manager/endpoint_v2/destination.py

@@ -229,7 +229,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()
 
@@ -871,9 +870,21 @@ def _push_to_queue(
             ).to_dict()
 
             queue_result_json = json.dumps(queue_result)
-            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")
+            # Get organization_id for proper HITL queue filtering
+            organization_id = UserContext.get_organization_identifier()
+            conn = QueueUtils.get_queue_inst(
+                {"use_hitl_backend": True, "organization_id": organization_id}
+            )
+            # API deployments don't have TTL, use system actor
+            conn.enqueue_with_ttl(
+                queue_name=q_name,
+                message=queue_result_json,
+                ttl_seconds=None,  # No TTL for API deployments
+                actor_id=None,  # System-initiated enqueue
+            )
+            logger.info(
+                f"Pushed {file_name} to queue {q_name} with organization_id {organization_id}"
+            )
             return
         connector_settings: dict[str, Any] = connector.connector_metadata
 
@@ -923,15 +934,22 @@ def _push_to_queue(
                 )
                 raise ValueError("Cannot enqueue empty JSON message")
 
-            conn = QueueUtils.get_queue_inst()
-
-            # Use the TTL metadata that was already set in the QueueResult object
+            # Get organization_id for proper HITL queue filtering
+            organization_id = UserContext.get_organization_identifier()
+            conn = QueueUtils.get_queue_inst(
+                {"use_hitl_backend": True, "organization_id": organization_id}
+            )
+            # Use TTL from workflow settings, system actor for workflow enqueue
             ttl_seconds = queue_result_obj.ttl_seconds
-
             conn.enqueue_with_ttl(
-                queue_name=q_name, message=queue_result_json, ttl_seconds=ttl_seconds
+                queue_name=q_name,
+                message=queue_result_json,
+                ttl_seconds=ttl_seconds,
+                actor_id=None,  # System-initiated enqueue
+            )
+            logger.info(
+                f"Pushed {file_name} to queue {q_name} with organization_id {organization_id} and TTL {ttl_seconds}"
             )
-            logger.info(f"Pushed {file_name} to queue {q_name} with file content")
 
     def _read_file_content_for_queue(self, input_file_path: str, file_name: str) -> str:
         """Read and encode file content for queue message.

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"