destination_connector.py

"""Destination Connector for Workflow Output Handling

This module provides specialized destination connector for handling workflow outputs,
extracted from the monolithic workflow_service.py to improve maintainability.

Handles:
- Filesystem destination output
- Database destination output
- API destination output
- Manual review queue output
- Output processing and validation
"""

import ast
import base64
import json
import os
import time
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Optional

from shared.enums import DestinationConfigKey, QueueResultStatus

# Import database utils (stable path)
from shared.infrastructure.database.utils import WorkerDatabaseUtils
from shared.infrastructure.logging import WorkerLogger
from shared.infrastructure.logging.helpers import log_file_error, log_file_info
from shared.models.result_models import QueueResult
from shared.utils.api_result_cache import get_api_cache_manager
from shared.utils.manual_review_factory import (
    get_manual_review_service,
    has_manual_review_plugin,
)
from shared.workflow.connectors.service import WorkerConnectorService
from shared.workflow.logger_helper import WorkflowLoggerHelper

from unstract.connectors.connectorkit import Connectorkit
from unstract.connectors.exceptions import ConnectorError
from unstract.core.data_models import ConnectionType as CoreConnectionType
from unstract.core.data_models import FileHashData
from unstract.core.exceptions import FileExecutionStageException
from unstract.core.file_execution_tracker import (
    FileExecutionStage,
    FileExecutionStageData,
    FileExecutionStageStatus,
    FileExecutionStatusTracker,
)
from unstract.filesystem import FileStorageType, FileSystem
from unstract.sdk1.constants import ToolExecKey
from unstract.sdk1.tool.mime_types import EXT_MIME_MAP
from unstract.workflow_execution.constants import (
    MetaDataKey,
    ToolMetadataKey,
    ToolOutputType,
)
from unstract.workflow_execution.execution_file_handler import (
    ExecutionFileHandler,
)

if TYPE_CHECKING:
    from ..api_client import InternalAPIClient

logger = WorkerLogger.get_logger(__name__)


@dataclass
class HandleOutputResult:
    """Result of handle_output method."""

    output: dict[str, Any] | str | None
    metadata: dict[str, Any] | None
    connection_type: str


@dataclass
class ExecutionContext:
    """Execution context for destination processing."""

    workflow_id: str
    execution_id: str
    organization_id: str
    file_execution_id: str
    api_client: Optional["InternalAPIClient"] = None
    workflow_log: Any = None


@dataclass
class FileContext:
    """File-specific context for processing."""

    file_hash: FileHashData
    file_name: str
    input_file_path: str
    workflow: dict[str, Any]
    execution_error: str | None = None


@dataclass
class HITLDecision:
    """Result of HITL routing decision."""

    should_route_to_hitl: bool = False
    reason: str | None = None  # Human-readable reason for the decision


@dataclass
class ProcessingResult:
    """Result of destination processing."""

    tool_execution_result: dict | str | None = None
    metadata: dict[str, Any] | None = None
    has_hitl: bool = False
    hitl_reason: str | None = None  # Reason why file was sent to HITL


@dataclass
class DestinationConfig:
    """Worker-compatible DestinationConfig implementation."""

    connection_type: str
    source_connection_type: str
    settings: dict[str, Any] = None
    is_api: bool = False
    use_file_history: bool = True
    # New connector instance fields from backend API
    connector_id: str | None = None
    connector_settings: dict[str, Any] = None
    connector_name: str | None = None
    # Manual review / HITL support
    hitl_queue_name: str | None = None
    hitl_packet_id: str | None = None
    # Source connector configuration for reading files
    source_connector_id: str | None = None
    source_connector_settings: dict[str, Any] = None
    file_execution_id: str | None = None

    def __post_init__(self):
        if self.settings is None:
            self.settings = {}
        if self.connector_settings is None:
            self.connector_settings = {}
        if self.source_connector_settings is None:
            self.source_connector_settings = {}
        # Determine if this is an API destination
        if self.connection_type and "api" in self.connection_type.lower():
            self.is_api = True

    def get_core_connection_type(self) -> CoreConnectionType:
        """Convert string connection_type to CoreConnectionType enum."""
        try:
            # Use the enum directly for consistent mapping
            connection_type_upper = self.connection_type.upper()

            # Try to get enum member by value
            for connection_type_enum in CoreConnectionType:
                if connection_type_enum.value == connection_type_upper:
                    return connection_type_enum

            # Fallback: handle legacy/unknown types
            logger.warning(
                f"Unknown connection type '{self.connection_type}', defaulting to DATABASE"
            )
            return CoreConnectionType.DATABASE

        except Exception as e:
            logger.error(
                f"Failed to convert connection type '{self.connection_type}' to enum: {e}"
            )
            return CoreConnectionType.DATABASE

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "DestinationConfig":
        """Create DestinationConfig from dictionary data."""
        connector_instance = data.get("connector_instance", {})
        return cls(
            connection_type=data.get("connection_type", ""),
            source_connection_type=data.get("source_connection_type"),
            settings=data.get("configuration", {}),
            use_file_history=data.get("use_file_history", True),
            connector_id=connector_instance.get("connector_id"),
            connector_settings=connector_instance.get("connector_metadata", {}),
            connector_name=connector_instance.get("connector_name"),
            hitl_queue_name=data.get("hitl_queue_name"),
            hitl_packet_id=data.get("hitl_packet_id"),
            source_connector_id=data.get("source_connector_id"),
            source_connector_settings=data.get("source_connector_settings", {}),
            file_execution_id=data.get("file_execution_id"),
        )


class WorkerDestinationConnector:
    """Worker-compatible destination connector following production patterns.

    This class replicates the functionality of backend DestinationConnector
    from workflow_manager/endpoint_v2/destination.py without Django dependencies.
    """

    # Use CoreConnectionType directly - no need for wrapper class

    def __init__(self, config: DestinationConfig, workflow_log=None):
        self.config = config
        self.connection_type = config.connection_type
        self.is_api = config.is_api
        self.use_file_history = config.use_file_history
        self.settings = config.settings
        self.workflow_log = workflow_log

        # Initialize logger helper for safe logging operations
        self.logger_helper = WorkflowLoggerHelper(workflow_log)

        # Store destination connector instance details
        self.connector_id = config.connector_id
        self.connector_settings = config.connector_settings
        self.connector_name = config.connector_name

        # Store source connector instance details for file reading
        self.source_connector_id = config.source_connector_id
        self.source_connector_settings = config.source_connector_settings

        # Manual review / HITL support
        self.hitl_queue_name = config.hitl_queue_name
        self.hitl_packet_id = config.hitl_packet_id

        # Workflow and execution context (will be set when handling output)
        self.organization_id = None
        self.workflow_id = None
        self.execution_id = None
        self.file_execution_id = None

        # Manual review service and API client (will be set when first needed)
        self.manual_review_service = None
        self._api_client = None

    @staticmethod
    def _extract_confidence_from_highlight_data(data: Any) -> float | None:
        """Extract confidence from 5th element of highlight data coordinate arrays.

        Recursively searches through nested arrays/objects to find coordinate arrays
        with 5 elements where the 5th element (index 4) is the confidence score.

        Args:
            data: Highlight data structure (can be nested arrays/dicts)

        Returns:
            Average confidence score if found, None otherwise
        """
        if not data:
            return None

        confidence_values = []

        def extract_from_array(arr):
            if isinstance(arr, list):
                for item in arr:
                    if isinstance(item, list):
                        # Check if this is a coordinate array with 5 elements
                        if len(item) >= 5 and isinstance(item[4], (int, float)):
                            confidence_values.append(float(item[4]))
                        else:
                            # Recursively check nested arrays
                            extract_from_array(item)
                    elif isinstance(item, dict):
                        # Recursively check objects
                        for val in item.values():
                            extract_from_array(val)
            elif isinstance(arr, dict):
                for val in arr.values():
                    extract_from_array(val)

        extract_from_array(data)

        # Calculate average confidence if we found any values
        if confidence_values:
            return sum(confidence_values) / len(confidence_values)

        return None

    def _prepare_result_for_rule_evaluation(
        self, file_name: str, tool_execution_result: dict | str | None
    ) -> dict | None:
        """Prepare result for rule evaluation by wrapping in expected structure.

        The rule engine expects: {"output": {...}, "metadata": {...}}

        Args:
            file_name: Name of the file (for logging)
            tool_execution_result: Raw tool execution result

        Returns:
            Wrapped result dict or None if no result
        """
        if not tool_execution_result:
            return None

        if not isinstance(tool_execution_result, dict):
            logger.warning(
                f"{file_name}: tool_execution_result is not a dict: {type(tool_execution_result)}"
            )
            return None

        # Check if tool_execution_result already has the correct structure
        if "output" in tool_execution_result:
            # Already in correct format with metadata - use directly
            return tool_execution_result

        # Legacy format: wrap in expected structure
        metadata = self.get_metadata()

        # 3-tier fallback hierarchy for confidence:
        # 1. word_confidence_data (if available)
        # 2. Extract from 5th element of highlight_data
        # 3. confidence_data (last resort)
        highlight_data = metadata.get("highlight_data", {}) if metadata else {}
        word_confidence_data = (
            metadata.get("word_confidence_data", {}) if metadata else {}
        )
        confidence_data = metadata.get("confidence_data", {}) if metadata else {}

        # If word_confidence_data is not available, try extracting from highlight_data
        if not word_confidence_data and highlight_data:
            extracted_confidence = self._extract_confidence_from_highlight_data(
                highlight_data
            )
            # Use extracted confidence if available, otherwise fall back to confidence_data
            if extracted_confidence is not None:
                # For rule engine, we provide a single confidence score
                confidence_data = {"_extracted_average": extracted_confidence}

        return {
            "output": tool_execution_result,
            "metadata": {
                "confidence_data": confidence_data,
                "word_confidence_data": word_confidence_data,
                "highlight_data": highlight_data,
                "whisper-hash": metadata.get("whisper-hash") if metadata else None,
                "extracted_text": metadata.get("extracted_text") if metadata else None,
            },
        }

    @classmethod
    def from_config(cls, workflow_log, config: DestinationConfig):
        """Create destination connector from config (matching Django backend interface)."""
        return cls(config, workflow_log)

    def _ensure_manual_review_service(
        self, api_client: Optional["InternalAPIClient"] = None
    ):
        """Ensure manual review service is initialized (lazy loading)."""
        if self.manual_review_service is None and api_client is not None:
            self._api_client = api_client
            self.manual_review_service = get_manual_review_service(
                api_client, api_client.organization_id
            )
        return self.manual_review_service

    def _get_destination_display_name(self) -> str:
        """Get human-readable destination name for logging."""
        if self.connection_type == CoreConnectionType.DATABASE.value:
            # Try to get database type from settings
            if self.connector_name:
                return f"database ({self.connector_name})"
            elif self.settings and "table" in self.settings:
                return f"database table '{self.settings['table']}'"
            return "database"
        elif self.connection_type == CoreConnectionType.FILESYSTEM.value:
            if self.connector_name:
                return f"filesystem ({self.connector_name})"
            return "filesystem destination"
        elif self.connection_type == CoreConnectionType.API.value:
            if self.connector_name:
                return f"API ({self.connector_name})"
            return "API endpoint"
        elif self.connection_type == CoreConnectionType.MANUALREVIEW.value:
            return "manual review queue"
        else:
            return f"{self.connection_type} destination"

    def _setup_execution_context(
        self,
        workflow_id: str,
        execution_id: str,
        organization_id: str,
        file_execution_id: str,
        api_client: Optional["InternalAPIClient"],
    ) -> ExecutionContext:
        """Setup and store execution context."""
        # Store in instance for backward compatibility with other methods
        self.workflow_id = workflow_id
        self.execution_id = execution_id
        self.organization_id = organization_id
        self.file_execution_id = file_execution_id

        return ExecutionContext(
            workflow_id=workflow_id,
            execution_id=execution_id,
            organization_id=organization_id,
            file_execution_id=file_execution_id,
            api_client=api_client,
            workflow_log=self.workflow_log,
        )

    def _setup_file_context(
        self,
        file_hash: FileHashData,
        workflow: dict[str, Any],
        execution_error: str | None,
    ) -> FileContext:
        """Setup file processing context."""
        return FileContext(
            file_hash=file_hash,
            file_name=file_hash.file_name,
            input_file_path=file_hash.file_path,
            workflow=workflow,
            execution_error=execution_error,
        )

    def _extract_processing_data(
        self, exec_ctx: ExecutionContext, file_ctx: FileContext
    ) -> ProcessingResult:
        """Extract tool results and metadata for processing."""
        tool_result = None
        if not file_ctx.execution_error:
            tool_result = self.get_tool_execution_result_from_execution_context(
                workflow_id=exec_ctx.workflow_id,
                execution_id=exec_ctx.execution_id,
                file_execution_id=exec_ctx.file_execution_id,
                organization_id=exec_ctx.organization_id,
            )

        metadata = self.get_metadata()

        return ProcessingResult(tool_execution_result=tool_result, metadata=metadata)

    def _check_and_acquire_destination_lock(
        self, exec_ctx: ExecutionContext, file_ctx: FileContext
    ) -> bool:
        """Check if destination already processed and atomically acquire lock using Redis SET NX.

        Returns:
            bool: True if lock acquired successfully, False if already processed (duplicate)

        This method provides duplicate prevention using Redis SET NX for atomic lock:
        1. Check if DESTINATION_PROCESSING, FINALIZATION, or COMPLETED stage already exists
        2. If yes, this is a duplicate attempt (e.g., from worker restart) -> skip
        3. If no, atomically acquire lock using Redis SET ... NX (single atomic operation)
        4. If lock acquisition succeeds, set DESTINATION_PROCESSING stage with longer TTL
        5. If lock acquisition fails, another worker has the lock -> skip
        """
        try:
            tracker = FileExecutionStatusTracker()

            # Get TTL values from environment
            LOCK_TTL = int(
                os.environ.get("DESTINATION_PROCESSING_LOCK_TTL_IN_SECOND", 120)
            )
            STAGE_TTL = int(
                os.environ.get("DESTINATION_PROCESSING_STAGE_TTL_IN_SECOND", 600)
            )

            # Get lock key for atomic operations
            lock_key = tracker.get_destination_lock_key(
                exec_ctx.execution_id, exec_ctx.file_execution_id
            )
            lock_token = str(uuid.uuid4())  # Unique token for debugging

            # STEP 1: Try to acquire lock atomically (SOURCE OF TRUTH)
            # This is atomic - if lock exists, another worker is processing
            logger.info(
                f"Attempting to acquire destination lock for file '{file_ctx.file_name}' "
                f"(lock_key={lock_key}, lock_ttl={LOCK_TTL}s, stage_ttl={STAGE_TTL}s)"
            )

            lock_acquired = tracker.redis_client.set(
                lock_key, lock_token, nx=True, ex=LOCK_TTL
            )

            # STEP 2: If lock acquisition failed, another worker is processing - WAIT
            if not lock_acquired:
                logger.info(
                    f"Lock already held by another worker for file '{file_ctx.file_name}'. "
                    f"Waiting for lock release to prevent premature chord cleanup (max {LOCK_TTL}s)..."
                )

                wait_start = time.time()
                max_wait = min(LOCK_TTL, 120)  # Wait up to lock TTL or 120s

                # Poll until lock released or timeout
                while time.time() - wait_start < max_wait:
                    # Check if lock released
                    if not tracker.redis_client.exists(lock_key):
                        wait_duration = time.time() - wait_start
                        # Lock released BEFORE timeout → Other worker finished (success or error) → Skip
                        logger.info(
                            f"Lock released for '{file_ctx.file_name}' after {wait_duration:.1f}s - "
                            f"other worker completed processing. Skipping as duplicate."
                        )
                        return False  # Skip immediately

                    time.sleep(2)  # Poll every 2 seconds

                # STEP 3: After wait, try to acquire lock again
                lock_acquired = tracker.redis_client.set(
                    lock_key, lock_token, nx=True, ex=LOCK_TTL
                )

                if not lock_acquired:
                    # Still can't acquire lock (timeout or another worker grabbed it)
                    if tracker.redis_client.exists(lock_key):
                        logger.warning(
                            f"Lock still held after {max_wait}s timeout for '{file_ctx.file_name}'. "
                            f"Another worker may have acquired it. Skipping as duplicate."
                        )
                    else:
                        logger.warning(
                            f"Failed to acquire lock for '{file_ctx.file_name}' after wait. "
                            f"Another worker may have grabbed it first. Skipping as duplicate."
                        )
                    return False  # Skip

            # Lock acquired successfully - now set DESTINATION_PROCESSING stage
            logger.info(
                f"Lock acquired successfully for file '{file_ctx.file_name}' (token={lock_token})"
            )

            try:
                tracker.update_stage_status(
                    exec_ctx.execution_id,
                    exec_ctx.file_execution_id,
                    FileExecutionStageData(
                        stage=FileExecutionStage.DESTINATION_PROCESSING,
                        status=FileExecutionStageStatus.IN_PROGRESS,
                    ),
                    ttl_in_second=STAGE_TTL,  # Use longer TTL for stage tracker
                )
                logger.info(
                    f"Successfully set DESTINATION_PROCESSING stage for file '{file_ctx.file_name}' "
                    f"with stage TTL {STAGE_TTL}s"
                )
                return True  # Lock acquired and stage set successfully

            except FileExecutionStageException:
                # Stage transition failed (shouldn't happen after lock acquired, but handle it)
                logger.exception(
                    f"Failed to set DESTINATION_PROCESSING stage after lock acquisition "
                    f"for file '{file_ctx.file_name}'. "
                    f"Releasing lock."
                )
                # Release the lock since we failed to set the stage
                tracker.redis_client.delete(lock_key)
                return False

        except Exception as e:
            # If Redis fails or other unexpected error, log but allow processing to continue
            # This ensures graceful degradation if tracking system is unavailable
            logger.exception(
                f"Failed to check/acquire destination lock for file '{file_ctx.file_name}': {e}. "
                f"Allowing processing to continue (graceful degradation)."
            )
            return True  # Allow processing on infrastructure failure

    def _check_and_handle_hitl(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ) -> HITLDecision:
        """Check HITL requirements and push to queue if needed.

        Returns:
            HITLDecision: Decision object with should_route_to_hitl flag and reason
        """
        hitl_decision = self._should_handle_hitl(
            file_name=file_ctx.file_name,
            file_hash=file_ctx.file_hash,
            workflow=file_ctx.workflow,
            api_client=exec_ctx.api_client,
            tool_execution_result=result.tool_execution_result,
            error=file_ctx.execution_error,
        )

        # Update result with HITL metadata
        result.has_hitl = hitl_decision.should_route_to_hitl
        result.hitl_reason = hitl_decision.reason

        if hitl_decision.should_route_to_hitl:
            self._push_data_to_queue(
                file_name=file_ctx.file_name,
                workflow=file_ctx.workflow,
                input_file_path=file_ctx.input_file_path,
                file_execution_id=exec_ctx.file_execution_id,
                tool_execution_result=result.tool_execution_result,
                api_client=exec_ctx.api_client,
                hitl_reason=hitl_decision.reason,
            )

        return hitl_decision

    def _process_destination(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ):
        """Route to appropriate destination handler."""
        handlers = {
            CoreConnectionType.API.value: self._handle_api_destination,
            CoreConnectionType.FILESYSTEM.value: self._handle_filesystem_destination,
            CoreConnectionType.DATABASE.value: self._handle_database_destination,
            CoreConnectionType.MANUALREVIEW.value: self._handle_manual_review_destination,
        }

        handler = handlers.get(self.connection_type)
        if handler:
            handler(exec_ctx, file_ctx, result)
        else:
            logger.warning(f"Unknown destination connection type: {self.connection_type}")

    def _handle_api_destination(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ):
        """Handle API destination processing."""
        log_file_info(
            exec_ctx.workflow_log,
            exec_ctx.file_execution_id,
            f"🔌 File '{file_ctx.file_name}' marked for API processing - preparing response",
        )

        # Enrich metadata with usage and pages processed data
        api_metadata = self.get_combined_metadata(exec_ctx.api_client, result.metadata)
        # Add HITL info only if plugin is available (cloud feature)
        if has_manual_review_plugin():
            api_metadata["hitl"] = {
                "file_sent_to_hitl": result.has_hitl,
                "reason": result.hitl_reason,
            }

        self.cache_api_result(
            api_client=exec_ctx.api_client,
            file_hash=file_ctx.file_hash,
            workflow_id=exec_ctx.workflow_id,
            execution_id=exec_ctx.execution_id,
            result=result.tool_execution_result if not result.has_hitl else None,
            file_execution_id=exec_ctx.file_execution_id,
            organization_id=exec_ctx.organization_id,
            error=file_ctx.execution_error,
            metadata=api_metadata,
        )

    def _handle_filesystem_destination(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ):
        """Handle filesystem destination processing."""
        if not result.has_hitl:
            if not result.tool_execution_result and not file_ctx.execution_error:
                error_msg = (
                    f"No tool execution result for file '{file_ctx.file_name}' "
                    f"- filesystem copy skipped"
                )
                logger.error(error_msg)
                log_file_error(
                    exec_ctx.workflow_log,
                    exec_ctx.file_execution_id,
                    f"❌ {error_msg}",
                )
                raise RuntimeError(error_msg)
            log_file_info(
                exec_ctx.workflow_log,
                exec_ctx.file_execution_id,
                f"📤 File '{file_ctx.file_name}' marked for FILESYSTEM processing - copying to destination",
            )
            self.copy_output_to_output_directory(
                file_ctx.input_file_path,
                exec_ctx.file_execution_id,
                exec_ctx.api_client,
            )
        else:
            logger.info(
                f"File '{file_ctx.file_name}' sent to HITL queue - FILESYSTEM processing will be handled after review"
            )

    def _handle_database_destination(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ):
        """Handle database destination processing."""
        if not result.has_hitl:
            log_file_info(
                exec_ctx.workflow_log,
                exec_ctx.file_execution_id,
                f"📤 File '{file_ctx.file_name}' marked for DATABASE processing - preparing to insert data",
            )
            if result.tool_execution_result or file_ctx.execution_error:
                self.insert_into_db(
                    file_ctx.input_file_path,
                    result.tool_execution_result,
                    result.metadata,
                    exec_ctx.file_execution_id,
                    error_message=file_ctx.execution_error,
                    api_client=exec_ctx.api_client,
                )
            else:
                error_msg = (
                    f"No tool execution result for file '{file_ctx.file_name}' "
                    f"- database insertion skipped"
                )
                logger.error(error_msg)
                log_file_error(
                    exec_ctx.workflow_log,
                    exec_ctx.file_execution_id,
                    f"❌ {error_msg}",
                )
                raise RuntimeError(error_msg)
        else:
            logger.info(
                f"File '{file_ctx.file_name}' sent to HITL queue - DATABASE processing will be handled after review"
            )

    def _handle_manual_review_destination(
        self,
        exec_ctx: ExecutionContext,
        file_ctx: FileContext,
        result: ProcessingResult,
    ):
        """Handle manual review destination processing."""
        log_file_info(
            exec_ctx.workflow_log,
            exec_ctx.file_execution_id,
            f"🔄 File '{file_ctx.file_name}' explicitly configured for MANUAL REVIEW - sending to queue",
        )

        if not result.has_hitl:
            self._push_data_to_queue(
                file_name=file_ctx.file_name,
                workflow=file_ctx.workflow,
                input_file_path=file_ctx.input_file_path,
                file_execution_id=exec_ctx.file_execution_id,
                tool_execution_result=result.tool_execution_result,
                api_client=exec_ctx.api_client,
                hitl_reason="Destination configured for manual review",
            )

    def _handle_destination_error(
        self, exec_ctx: ExecutionContext, file_ctx: FileContext, error: Exception
    ):
        """Handle destination processing errors."""
        logger.error(f"Destination handle_output failed: {str(error)}")
        log_file_error(
            exec_ctx.workflow_log,
            exec_ctx.file_execution_id,
            f"❌ File '{file_ctx.file_name}' failed to send to destination: {str(error)}",
        )

    def _log_processing_success(
        self, exec_ctx: ExecutionContext, file_ctx: FileContext, has_hitl: bool
    ):
        """Log successful processing."""
        if has_hitl:
            destination_name = "HITL/MANUAL REVIEW"
        else:
            destination_name = self._get_destination_display_name()
        log_file_info(
            exec_ctx.workflow_log,
            exec_ctx.file_execution_id,
            f"✅ File '{file_ctx.file_name}' successfully sent to {destination_name}",
        )

    def cache_api_result(
        self,
        file_hash: FileHashData,
        workflow_id: str,
        execution_id: str,
        file_execution_id: str,
        organization_id: str,
        api_client: Any | None,
        # file_history: dict[str, Any] | None,
        result: dict[str, Any] | None,
        error: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> bool:
        """Cache API result using APIResultCacheManager."""
        try:
            # Calculate accurate elapsed time from workflow start time
            if metadata and MetaDataKey.WORKFLOW_START_TIME in metadata:
                workflow_start_time = metadata[MetaDataKey.WORKFLOW_START_TIME]
                current_time = time.time()
                actual_elapsed_time = current_time - workflow_start_time

                # Update total_elapsed_time with accurate measurement
                metadata[MetaDataKey.TOTAL_ELAPSED_TIME] = actual_elapsed_time

                logger.info(
                    f"TIMING: Calculated accurate elapsed time for API caching: {actual_elapsed_time:.3f}s "
                    f"(from workflow start: {workflow_start_time:.6f} to now: {current_time:.6f})"
                )

            # Use APIResultCacheManager for consistent caching behavior
            api_cache_manager = get_api_cache_manager()
            success = api_cache_manager.cache_api_result_direct(
                file_name=file_hash.file_name,
                file_execution_id=file_execution_id,
                workflow_id=workflow_id,
                execution_id=execution_id,
                result=result,
                error=error,
                organization_id=organization_id,
                metadata=metadata,
            )

            if success:
                logger.info(
                    f"Successfully cached API result for execution {execution_id}"
                )
            else:
                logger.warning(f"Failed to cache API result for execution {execution_id}")

            return success

        except Exception as e:
            logger.error(
                f"Failed to cache API result for execution {execution_id}: {str(e)}"
            )
            # Return False but don't re-raise - caching failures shouldn't stop execution
            raise

    def handle_output(
        self,
        is_success: bool,
        file_hash: FileHashData,
        # file_history: dict[str, Any] | None,
        workflow: dict[str, Any],
        file_execution_id: str = None,
        api_client: Optional["InternalAPIClient"] = None,
        workflow_id: str = None,
        execution_id: str = None,
        organization_id: str = None,
        execution_error: str = None,
    ) -> HandleOutputResult | None:
        """Handle the output based on the connection type.

        This refactored version uses clean architecture with context objects
        and single-responsibility methods for better maintainability.
        """
        # Setup contexts
        exec_ctx = self._setup_execution_context(
            workflow_id, execution_id, organization_id, file_execution_id, api_client
        )
        file_ctx = self._setup_file_context(file_hash, workflow, execution_error)

        # Log if HITL queue is configured (reduced debug logging)
        if self.hitl_queue_name:
            logger.debug(f"HITL queue configured: {self.hitl_queue_name}")

        # Check if destination already processed and atomically acquire lock FIRST
        # This prevents duplicate insertions during warm shutdown scenarios
        # IMPORTANT: Check lock BEFORE extracting data to avoid unnecessary work
        lock_acquired = self._check_and_acquire_destination_lock(exec_ctx, file_ctx)
        if not lock_acquired:
            # Duplicate detected or another worker has the lock - abort ALL processing
            logger.info(
                f"Duplicate detected for file '{file_ctx.file_name}' - "
                f"aborting ALL processing (lock not acquired, already processed or being processed by another worker)"
            )
            # Return None to signal to caller that this is a duplicate skip
            # Caller should not create file history, update stages, or clean up locks
            return None

        # Only extract data if lock was acquired (not a duplicate)
        result = self._extract_processing_data(exec_ctx, file_ctx)

        # Check and handle HITL if needed (this updates result.has_hitl and result.hitl_reason)
        self._check_and_handle_hitl(exec_ctx, file_ctx, result)

        # Process through appropriate destination
        try:
            self._process_destination(exec_ctx, file_ctx, result)
        except Exception as e:
            self._handle_destination_error(exec_ctx, file_ctx, e)
            raise
        finally:
            # Release lock after destination processing completes
            # Critical section (stage set + data extraction + destination write) is done
            # File history and stage updates don't need the lock (protected by stage checks)
            try:
                tracker = FileExecutionStatusTracker()
                lock_key = tracker.get_destination_lock_key(
                    exec_ctx.execution_id, exec_ctx.file_execution_id
                )
                tracker.redis_client.delete(lock_key)
                logger.info(
                    f"Released destination lock for '{file_ctx.file_name}' "
                    f"after destination processing (lock_key={lock_key})"
                )
            except Exception as lock_error:
                logger.warning(
                    f"Failed to release destination lock for '{file_ctx.file_name}': {lock_error}"
                )

        # Log success
        self._log_processing_success(exec_ctx, file_ctx, result.has_hitl)

        return HandleOutputResult(
            output=result.tool_execution_result,
            metadata=result.metadata,
            connection_type=self.connection_type,
        )

    def get_combined_metadata(
        self, api_client: "InternalAPIClient", metadata: dict[str, Any] = None
    ) -> dict[str, Any]:
        """Get combined workflow and usage metadata.

        Returns:
            dict[str, Any]: Combined metadata including workflow and usage data.
        """
        if metadata is None:
            metadata = {}
        if not api_client:
            return metadata
        file_execution_id = self.file_execution_id
        usage_metadata = api_client.get_aggregated_token_count(file_execution_id)

        if usage_metadata:
            metadata["usage"] = usage_metadata.to_dict()
        if file_execution_id:
            metadata["total_pages_processed"] = api_client.get_aggregated_pages_processed(
                file_execution_id
            )
        return metadata

    def insert_into_db(
        self,
        input_file_path: str,
        tool_execution_result: str = None,
        metadata: dict[str, Any] = None,
        file_execution_id: str = None,
        error_message: str = None,
        api_client: "InternalAPIClient" = None,
    ) -> None:
        """Insert data into the database (following production pattern)."""
        # If no data and no error, don't execute CREATE or INSERT query
        if not (tool_execution_result or error_message):
            raise ValueError("No tool_execution_result or error_message provided")

        if error_message:
            logger.info(
                f"Proceeding with error record insertion for {input_file_path}: {error_message}"
            )

        # Store file_execution_id for logging
        if file_execution_id:
            self.current_file_execution_id = file_execution_id

        # Extract connector instance details from instance variables (now properly set)
        connector_id = self.connector_id
        connector_settings = self.connector_settings

        logger.info(f"Database destination - Connector ID: {connector_id}")
        logger.info(
            f"Database destination - Connector settings available: {bool(connector_settings)}"
        )
        logger.info(
            f"Database destination - Settings keys: {list(self.settings.keys()) if self.settings else 'None'}"
        )

        if not connector_id:
            raise ValueError("No connector_id provided in destination configuration")

        if not connector_settings:
            raise ValueError(
                "No connector_settings provided in destination configuration"
            )

        db_class = WorkerDatabaseUtils.get_db_class(
            connector_id=connector_id,
            connector_settings=connector_settings,
        )

        # Get combined metadata including usage data
        metadata = self.get_combined_metadata(api_client, metadata)
        logger.info(f"Database destination - Metadata: {metadata}")

        # Get table configuration from destination settings (table-specific config)
        table_name = str(self.settings.get("table", "unstract_results"))
        include_agent = bool(self.settings.get("includeAgent", False))
        include_timestamp = bool(self.settings.get("includeTimestamp", False))
        agent_name = str(self.settings.get("agentName", "UNSTRACT_DBWRITER"))
        column_mode = str(
            self.settings.get("columnMode", "WRITE_JSON_TO_A_SINGLE_COLUMN")
        )
        single_column_name = str(self.settings.get("singleColumnName", "data"))
        file_path_name = str(self.settings.get("filePath", "file_path"))
        execution_id_name = str(self.settings.get("executionId", "execution_id"))

        # Get tool execution result (use provided result only)
        data = tool_execution_result

        # Remove metadata from result
        # Tool text-extractor returns data in the form of string.
        # Don't pop out metadata in this case.
        if isinstance(data, dict):
            data.pop("metadata", None)

        # Use the workflow execution ID - warn if not available
        if not self.execution_id:
            logger.warning("Workflow execution_id not provided, using NULL in database")
            execution_id = None
        else:
            execution_id = self.execution_id

        engine = None
        try:
            logger.info(f"Creating database connection with connector ID: {connector_id}")
            db_class = WorkerDatabaseUtils.get_db_class(
                connector_id=connector_id,
                connector_settings=connector_settings,
            )
            table_info = db_class.get_information_schema(table_name=table_name)

            logger.info(
                f"destination connector table_name: {table_name} with table_info: {table_info}"
            )
            engine = db_class.get_engine()

            if table_info:
                if db_class.has_no_metadata(table_info=table_info):
                    table_info = WorkerDatabaseUtils.migrate_table_to_v2(
                        db_class=db_class,
                        engine=engine,
                        table_name=table_name,
                        column_name=single_column_name,
                    )

            logger.info(f"Creating table {table_name} if not exists")

            values = WorkerDatabaseUtils.get_columns_and_values(
                column_mode_str=column_mode,
                data=data,
                include_timestamp=include_timestamp,
                include_agent=include_agent,
                agent_name=agent_name,
                single_column_name=single_column_name,
                file_path_name=file_path_name,
                execution_id_name=execution_id_name,
                file_path=input_file_path,
                execution_id=execution_id,
                metadata=metadata,
                error=error_message,
            )

            WorkerDatabaseUtils.create_table_if_not_exists(
                db_class=db_class,
                engine=engine,
                table_name=table_name,
                database_entry=values,
            )

            # Remove None values from INSERT to let database handle as NULL
            # Table schema already created with all columns (including data column)
            # Removing None values prevents "invalid JSON" errors when inserting error records
            values = {k: v for k, v in values.items() if v is not None}

            logger.info(f"Preparing SQL query data for table {table_name}")
            sql_columns_and_values = WorkerDatabaseUtils.get_sql_query_data(
                conn_cls=db_class,
                table_name=table_name,
                values=values,
            )
            logger.info(
                f"sql_columns_and_values for table_name: {table_name} are: {sql_columns_and_values}"
            )
            logger.info(
                f"Executing insert query for {len(sql_columns_and_values)} columns"
            )
            WorkerDatabaseUtils.execute_write_query(
                db_class=db_class,
                engine=engine,
                table_name=table_name,
                sql_keys=list(sql_columns_and_values.keys()),
                sql_values=list(sql_columns_and_values.values()),
            )

            logger.info(f"Successfully inserted data into database table {table_name}")

            # Log to UI with file_execution_id for better correlation
            if hasattr(self, "current_file_execution_id"):
                log_file_info(
                    self.workflow_log,
                    self.current_file_execution_id,
                    f"📥 Data successfully inserted into database table '{table_name}'",
                )
        except ConnectorError as e:
            error_msg = f"Database connection failed for {input_file_path}: {str(e)}"
            logger.error(error_msg)
            raise
        except Exception as e:
            error_msg = (
                f"Failed to insert data into database for {input_file_path}: {str(e)}"
            )
            logger.error(error_msg)
            raise
        finally:
            self._close_engine(engine, input_file_path)

    def _close_engine(self, engine: Any, input_file_path: str) -> None:
        """Safely close database engine."""
        if engine:
            try:
                engine.close()
            except Exception as e:
                logger.error(
                    f"Failed to close database engine for {input_file_path}: {str(e)}"
                )

    def copy_output_to_output_directory(
        self,
        input_file_path: str,
        file_execution_id: str = None,
        api_client: Optional["InternalAPIClient"] = None,
    ) -> None:
        """Copy output to the destination directory (following backend production pattern).

        This method should ONLY be called for TASK workflows where destination is FILESYSTEM.
        ETL workflows (destination=DATABASE) use insert_into_db() instead.
        API workflows don't use this destination copying logic.
        """
        # Store file_execution_id for logging
        if file_execution_id:
            self.current_file_execution_id = file_execution_id

        # ARCHITECTURE VALIDATION: Ensure this is only called for TASK workflows
        if api_client and self.workflow_id:
            if self.connection_type != CoreConnectionType.FILESYSTEM.value:
                logger.warning(
                    f"copy_output_to_output_directory called for destination connection_type {self.connection_type} with workflow {self.workflow_id} - this should only be used for {CoreConnectionType.FILESYSTEM.value} workflows"
                )
                return

        # Copy output to filesystem destination

        try:
            # Get destination connector settings - exactly like backend
            if not self.connector_id or not self.connector_settings:
                raise ValueError(
                    f"Missing destination connector configuration: connector_id={self.connector_id}, "
                    f"settings={bool(self.connector_settings)}"
                )

            # Import necessary modules for file operations

            # Get connector settings and configuration like backend (lines 262-265)
            connector_settings = self.connector_settings
            destination_configurations = self.settings or {}

            # Extract destination path configuration using enums to prevent camelCase/snake_case issues
            root_path = str(connector_settings.get(DestinationConfigKey.PATH, ""))
            output_directory = str(
                destination_configurations.get(DestinationConfigKey.OUTPUT_FOLDER, "/")
            )

            # Get the destination connector instance (lines 270-272)
            logger.debug(f"Initializing destination connector: {self.connector_id}")
            connector_service = WorkerConnectorService(api_client)
            destination_fs = connector_service._get_destination_connector(
                connector_id=self.connector_id, connector_settings=connector_settings
            )

            # Get connector root directory like backend (lines 273-275)
            output_directory = destination_fs.get_connector_root_dir(
                input_dir=output_directory, root_path=root_path
            )

            # Build destination volume path like backend (lines 277-279)
            # Backend uses self.file_execution_dir which maps to our execution path
            execution_dir_path = f"unstract/execution/{self.organization_id}/{self.workflow_id}/{self.execution_id}/{file_execution_id}"
            destination_volume_path = os.path.join(
                execution_dir_path, ToolExecKey.OUTPUT_DIR
            )

            # Get workflow execution file system for reading (like backend lines 285-286)
            file_system = FileSystem(FileStorageType.WORKFLOW_EXECUTION)
            fs = file_system.get_file_storage()

            # Backend logic: Create destination directory if needed (line 282)
            try:
                # CONNECTOR COMPATIBILITY: Skip root directory creation for certain paths
                normalized_output_dir = output_directory.strip("/")
                if normalized_output_dir and normalized_output_dir != ".":
                    destination_fs.create_dir_if_not_exists(
                        input_dir=normalized_output_dir
                    )
                else:
                    logger.debug(
                        f"Skipping root directory creation for path: '{output_directory}'"
                    )
            except Exception as e:
                logger.warning(
                    f"Could not create destination directory {output_directory}: {e}"
                )

            # Backend logic: Walk the OUTPUT_DIR and copy everything (lines 287-307)
            copied_files = []
            failed_files = []
            total_copied = 0

            # Check if OUTPUT_DIR exists before walking
            if not fs.exists(destination_volume_path):
                logger.warning(
                    f"Output directory does not exist: {destination_volume_path}"
                )
                logger.info(
                    "No output files to copy - workflow may not have produced output"
                )
            else:
                # Walk directory structure like backend (lines 289-307)
                try:
                    dir_path = fs.walk(str(destination_volume_path))

                    for root, dirs, files in dir_path:
                        # Create directories in destination (lines 290-296)
                        for dir_name in dirs:
                            current_dir = os.path.join(
                                output_directory,
                                os.path.relpath(root, destination_volume_path),
                                dir_name,
                            )
                            try:
                                # CONNECTOR COMPATIBILITY: Skip root directory creation for certain paths
                                normalized_current_dir = current_dir.strip("/")
                                if (
                                    normalized_current_dir
                                    and normalized_current_dir != "."
                                ):
                                    destination_fs.create_dir_if_not_exists(
                                        input_dir=normalized_current_dir
                                    )
                                    logger.debug(
                                        f"Created directory: {normalized_current_dir}"
                                    )
                                else:
                                    logger.debug(
                                        f"Skipping root directory creation for path: '{current_dir}'"
                                    )
                            except Exception as e:
                                logger.warning(
                                    f"Could not create directory {current_dir}: {e}"
                                )

                        # Copy files (lines 298-307)
                        for file_name in files:
                            source_path = os.path.join(root, file_name)

                            # Calculate relative path and handle path construction properly
                            relative_path = os.path.relpath(root, destination_volume_path)

                            if relative_path == "." or not relative_path:
                                # When root == destination_volume_path, use output_directory directly
                                destination_path = os.path.join(
                                    output_directory, file_name
                                )
                            else:
                                destination_path = os.path.join(
                                    output_directory,
                                    relative_path,
                                    file_name,
                                )

                            # Normalize path and validate
                            destination_path = os.path.normpath(destination_path)

                            # ARCHITECTURE FIX: Proper error handling instead of inconsistent fallbacks
                            if not destination_path or destination_path in [".", "/"]:
                                error_msg = f"Invalid destination path '{destination_path}' constructed for file {file_name}"
                                logger.error(f"ERROR: {error_msg}")
                                logger.error(
                                    f"ERROR: Debug info - output_directory='{output_directory}', relative_path='{relative_path}', root='{root}', destination_volume_path='{destination_volume_path}'"
                                )
                                raise ValueError(error_msg)

                            try:
                                # CONNECTOR COMPATIBILITY: Handle path normalization for worker context
                                # Remove leading slash that can cause issues with various connectors
                                final_destination_path = (
                                    destination_path.lstrip("/")
                                    if destination_path.startswith("/")
                                    else destination_path
                                )

                                # Validate the final path is not empty after normalization
                                if (
                                    not final_destination_path
                                    or final_destination_path in [".", ""]
                                ):
                                    raise ValueError(
                                        f"Invalid destination path after normalization: '{final_destination_path}' (original: '{destination_path}')"
                                    )

                                destination_fs.upload_file_to_storage(
                                    source_path=source_path,
                                    destination_path=final_destination_path,
                                )
                                copied_files.append(file_name)
                                total_copied += 1
                                logger.debug(f"✅ Successfully copied: {file_name}")

                            except Exception as copy_error:
                                logger.error(
                                    f"Failed to copy {file_name}: {copy_error}",
                                    exc_info=True,
                                )
                                failed_files.append(file_name)
                                # Continue with other files even if one fails

                except Exception as walk_error:
                    logger.error(
                        f"Failed to walk output directory {destination_volume_path}: {walk_error}"
                    )

            # Report results - handle both successes and failures
            if failed_files:
                # If any files failed, this should be treated as an error
                failed_count = len(failed_files)
                if failed_count == 1:
                    error_message = (
                        f"❌ Failed to copy file to destination: {failed_files[0]}"
                    )
                else:
                    error_message = (
                        f"❌ Failed to copy {failed_count} files to destination"
                    )
                logger.error(error_message)

                # Log to UI with file_execution_id
                if hasattr(self, "current_file_execution_id"):
                    log_file_info(
                        self.workflow_log,
                        self.current_file_execution_id,
                        error_message,
                    )

                # Raise exception to trigger proper error handling
                raise Exception(f"Destination copy failed: {error_message}")
            elif total_copied > 0:
                success_message = f"💾 Successfully copied {total_copied} files to filesystem destination"
                logger.info(success_message)

                # Log to UI
                if hasattr(self, "current_file_execution_id"):
                    log_file_info(
                        self.workflow_log,
                        self.current_file_execution_id,
                        success_message,
                    )
            else:
                success_message = (
                    "💾 No output files found to copy to filesystem destination"
                )
                logger.info(success_message)

                # Log to UI
                if hasattr(self, "current_file_execution_id"):
                    log_file_info(
                        self.workflow_log,
                        self.current_file_execution_id,
                        success_message,
                    )

        except Exception as e:
            error_msg = f"Failed to copy files to filesystem destination: {str(e)}"
            logger.error(error_msg, exc_info=True)

            # Log error to UI
            if hasattr(self, "current_file_execution_id"):
                log_file_info(
                    self.workflow_log,
                    self.current_file_execution_id,
                    f"❌ {error_msg}",
                )

            # Re-raise the exception so that destination processing can fail properly
            raise Exception(f"Destination filesystem copy failed: {str(e)}") from e

    def get_tool_execution_result(
        self, file_history=None, tool_execution_result: str = None
    ) -> Any:
        """Get result data from the output file (following production pattern)."""
        if tool_execution_result:
            return tool_execution_result

        if file_history and hasattr(file_history, "result") and file_history.result:
            return self.parse_string(file_history.result)

        # Default fallback - could be enhanced to read from actual execution files
        return None

    def get_tool_execution_result_from_execution_context(
        self,
        workflow_id: str,
        execution_id: str,
        file_execution_id: str,
        organization_id: str,
    ) -> dict | str:
        """Get tool execution result using proper execution context (preferred method)."""
        try:
            # Use ExecutionFileHandler to get proper paths (matching backend)
            file_handler = ExecutionFileHandler(
                workflow_id=workflow_id,
                execution_id=execution_id,
                organization_id=organization_id,
                file_execution_id=file_execution_id,
            )
            metadata_file_path = file_handler.metadata_file
            # Get workflow metadata (following backend pattern)
            file_system = FileSystem(FileStorageType.WORKFLOW_EXECUTION)
            file_storage = file_system.get_file_storage()

            if not metadata_file_path:
                logger.warning(
                    "No metadata_file_path (file_execution_id=%s, execution_dir=%s)",
                    file_execution_id,
                    file_handler.execution_dir,
                )
                return None

            if not file_storage.exists(metadata_file_path):
                logger.warning(
                    "METADATA.json not found at '%s'",
                    metadata_file_path,
                )
                return None

            metadata_content = file_storage.read(path=metadata_file_path, mode="r")
            metadata = json.loads(metadata_content)

            # Get output type from metadata (following backend pattern)
            output_type = self._get_output_type_from_metadata(metadata)

            # Get the output file path using the file handler (matching backend pattern)
            output_file_path = file_handler.infile

            if not output_file_path:
                logger.warning(
                    "No infile path (file_execution_id=%s)",
                    file_execution_id,
                )
                return None

            if not file_storage.exists(output_file_path):
                logger.warning(
                    "INFILE not found at '%s'",
                    output_file_path,
                )
                return None

            file_type = file_storage.mime_type(path=output_file_path)
            # Note: INFILE has no extension, so mime_type may return text/plain
            # Allow text/plain for both JSON and TXT since we validate content anyway
            allowed_mime_types = [
                EXT_MIME_MAP[ToolOutputType.JSON.lower()],
                EXT_MIME_MAP[ToolOutputType.TXT.lower()],
                "text/plain",  # INFILE has no extension
            ]

            if output_type == ToolOutputType.JSON:
                if file_type not in allowed_mime_types:
                    msg = f"Expected tool output type: JSON, got: '{file_type}'"
                    logger.error(msg)
                    raise Exception(msg)
                file_content = file_storage.read(output_file_path, mode="r")
                result = json.loads(file_content)
                return result
            elif output_type == ToolOutputType.TXT:
                if file_type not in allowed_mime_types:
                    msg = f"Expected tool output type: TXT, got: '{file_type}'"
                    logger.error(msg)
                    raise Exception(msg)
                file_content = file_storage.read(output_file_path, mode="r")
                result = file_content.encode("utf-8").decode("unicode-escape")
                return result
            else:
                raise Exception(f"Unsupported output type: {output_type}")

        except Exception as e:
            logger.error(
                f"Exception while getting tool execution result from execution context: {str(e)}",
                exc_info=True,
            )
            raise

    def _get_output_type_from_metadata(self, metadata: dict[str, Any]) -> str:
        """Get output type from metadata (following backend pattern)."""
        try:
            # Get tool metadata list
            tool_metadata = metadata.get(MetaDataKey.TOOL_METADATA, [])
            if not tool_metadata:
                logger.warning("No tool metadata found, defaulting to TXT output")
                return ToolOutputType.TXT

            # Get last tool metadata (like backend)
            last_tool_metadata = tool_metadata[-1]
            output_type = last_tool_metadata.get(
                ToolMetadataKey.OUTPUT_TYPE, ToolOutputType.TXT
            )

            logger.debug(f"Detected output type: {output_type}")
            return output_type

        except Exception as e:
            logger.error(f"Failed to get output type from metadata: {str(e)}")
            return ToolOutputType.TXT

    def parse_string(self, original_string: str) -> Any:
        """Parse the given string, attempting to evaluate it as a Python literal."""
        try:
            # Try to evaluate as a Python literal
            python_literal = ast.literal_eval(original_string)
            return python_literal
        except (SyntaxError, ValueError):
            # If evaluating as a Python literal fails,
            # assume it's a plain string
            return original_string

    def get_metadata(self, file_history=None) -> dict[str, Any] | None:
        """Get metadata from the output file (matching backend pattern).

        This matches backend DestinationConnector.get_metadata.
        """
        if file_history:
            if self.has_valid_metadata(getattr(file_history, "metadata", None)):
                return self.parse_string(file_history.metadata)
            else:
                return None
        metadata: dict[str, Any] = self._get_workflow_metadata()
        return metadata

    def _get_workflow_metadata(self) -> dict[str, Any]:
        """Get metadata from the workflow (matching backend pattern)."""
        file_handler = ExecutionFileHandler(
            workflow_id=self.workflow_id,
            execution_id=self.execution_id,
            organization_id=self.organization_id,
            file_execution_id=self.file_execution_id,
        )
        metadata: dict[str, Any] = file_handler.get_workflow_metadata()
        return metadata

    def has_valid_metadata(self, metadata: Any) -> bool:
        """Check if metadata is valid (matching backend pattern)."""
        # Check if metadata is not None and metadata is a non-empty string
        if not metadata:
            return False
        if not isinstance(metadata, str):
            return False
        if metadata.strip().lower() == "none":
            return False
        return True

    def _should_handle_hitl(
        self,
        file_name: str,
        file_hash: FileHashData,
        workflow: dict[str, Any],
        api_client: Optional["InternalAPIClient"] = None,
        tool_execution_result: dict | str | None = None,
        error: str | None = None,
    ) -> HITLDecision:
        """Determines if HITL processing should be performed.

        This method replicates the backend DestinationConnector._should_handle_hitl logic.

        The logic is:
        1. hitl_packet_id takes precedence - always push to packet queue
        2. hitl_queue_name with API rules:
           - If no API rules configured → Push ALL to HITL (backward compatible)
           - If API rules exist → Evaluate rules, push only if rules match
        3. Database destination → Evaluate DB rules

        Returns:
            HITLDecision: Decision object with should_route_to_hitl flag and reason
        """
        logger.info(f"{file_name}: checking if file is eligible for HITL")
        if error:
            logger.error(
                f"{file_name}: file is not eligible for HITL due to error: {error}"
            )
            return HITLDecision(False, f"Processing error: {error}")

        # Check hitl_packet_id first - it takes precedence over everything else
        if self.hitl_packet_id:
            logger.info(
                f"API packet override: pushing to packet queue for file {file_name}"
            )
            return HITLDecision(True, f"HITL packet ID specified: {self.hitl_packet_id}")

        # Check if API deployment requested HITL override with hitl_queue_name
        if self.hitl_queue_name:
            logger.info(
                f"{file_name}: HITL queue name detected: {self.hitl_queue_name}. "
                f"Checking for API rules..."
            )
            # Get workflow util to check for API rules
            manual_review_service = self._ensure_manual_review_service(api_client)
            if manual_review_service:
                workflow_util = manual_review_service.get_workflow_util()
                has_rules = workflow_util.has_api_rules(workflow)
                logger.info(f"{file_name}: has_api_rules={has_rules}")

                if has_rules:
                    # API rules exist - evaluate them
                    wrapped_result = self._prepare_result_for_rule_evaluation(
                        file_name, tool_execution_result
                    )
                    logger.info(
                        f"{file_name}: Evaluating API rules. "
                        f"wrapped_result type={type(wrapped_result).__name__}, "
                        f"has_output={bool(wrapped_result.get('output') if wrapped_result else False)}"
                    )
                    # Get detailed rule evaluation result
                    rule_result = workflow_util.validate_rule_engine_with_reason(
                        wrapped_result,
                        workflow,
                        file_hash.file_destination,
                        file_hash.is_manualreview_required,
                        rule_type="API",
                    )
                    if rule_result.get("matched", False):
                        reason = rule_result.get("reason", "API rules matched")
                        logger.info(
                            f"{file_name}: API rules matched - pushing to HITL queue. Reason: {reason}"
                        )
                        return HITLDecision(True, reason)
                    else:
                        reason = rule_result.get("reason", "API rules not matched")
                        logger.info(
                            f"{file_name}: API rules not matched - returning result directly. Reason: {reason}"
                        )
                        return HITLDecision(False, reason)
            else:
                logger.warning(f"{file_name}: No manual review service available")

            # No API rules configured - backward compatible behavior: push ALL to HITL
            logger.info(
                f"{file_name}: No API rules configured - pushing ALL to HITL queue (backward compatible)"
            )
            return HITLDecision(
                True, "HITL queue specified, no rules configured - sending all to HITL"
            )

        # Skip HITL validation if we're using file_history and no execution result is available
        # CRITICAL FIX: Only skip HITL validation for API deployments without file history
        # ETL workflows should ALWAYS evaluate rules, regardless of file_history setting
        if self.is_api and not self.use_file_history:
            logger.info(
                f"{file_name}: Skipping HITL validation for API deployment without file history"
            )
            return HITLDecision(
                False, "API deployment without file history - HITL not applicable"
            )

        # For API deployments WITH file history, continue to evaluate rules
        # For ETL workflows, ALWAYS evaluate rules (regardless of file_history)

        # Use class-level manual review service
        manual_review_service = self._ensure_manual_review_service(api_client)
        if not manual_review_service:
            logger.info(f"No manual review service available for {file_name}")
            return HITLDecision(False, "Manual review service not available")

        workflow_util = manual_review_service.get_workflow_util()

        # Don't skip rule evaluation just because file wasn't pre-selected by percentage
        # validate_rule_engine will check both percentage selection AND rules with OR/AND logic
        if not file_hash.is_manualreview_required:
            logger.info(
                f"{file_name}: File not pre-selected by percentage, "
                f"but checking if rules match..."
            )

        # Prepare result for rule evaluation using helper method
        wrapped_result = self._prepare_result_for_rule_evaluation(
            file_name, tool_execution_result
        )

        # Evaluate DB rules for ETL workflows with reason
        rule_result = workflow_util.validate_rule_engine_with_reason(
            wrapped_result,
            workflow,
            file_hash.file_destination,
            file_hash.is_manualreview_required,
            rule_type="DB",
        )
        is_to_hitl = rule_result.get("matched", False)
        reason = rule_result.get("reason", "DB rules evaluation")
        logger.info(
            f"File {file_name} checked for manual review: {is_to_hitl}, reason: {reason}"
        )
        return HITLDecision(is_to_hitl, reason)

    def _enqueue_to_packet_or_regular_queue(
        self,
        file_name: str,
        queue_result: dict[str, Any],
        queue_name: str,
        workflow_util: Any,
        ttl_seconds: int | None = None,
    ) -> None:
        """Route to packet queue or regular queue based on hitl_packet_id.

        Args:
            file_name: Name of the file being queued
            queue_result: Queue result dictionary
            queue_name: Queue name for regular queue
            workflow_util: Workflow utility instance for queue operations
            ttl_seconds: TTL in seconds (optional, for regular queue)
        """
        if self.hitl_packet_id:
            # Route to packet queue via backend API (enterprise only)
            logger.info(f"Routing {file_name} to packet queue {self.hitl_packet_id}")

            # Access the manual review client from workflow_util
            # Enterprise: workflow_util.client is ManualReviewAPIClient
            # OSS: workflow_util is null implementation without client attribute
            manual_review_client = getattr(workflow_util, "client", None)

            # Check if client exists and has enqueue_to_packet method (enterprise only)
            if not manual_review_client or not hasattr(
                manual_review_client, "enqueue_to_packet"
            ):
                error_msg = (
                    f"Packet queues are not available"
                    f"Cannot enqueue file '{file_name}' to packet '{self.hitl_packet_id}'. "
                    f"This is an Enterprise-only feature."
                )
                logger.error(error_msg)
                raise NotImplementedError(error_msg)

            # Call backend API to enqueue to packet queue
            result = manual_review_client.enqueue_to_packet(
                hitl_packet_id=self.hitl_packet_id,
                queue_result=queue_result,
                organization_id=self.organization_id,
            )

            if not result.get("success", False):
                error_msg = (
                    f"Failed to push {file_name} to packet {self.hitl_packet_id}: "
                    f"{result.get('message', 'Unknown error')}"
                )
                logger.error(error_msg)
                raise RuntimeError(error_msg)

            logger.info(
                f"✅ MANUAL REVIEW: File '{file_name}' sent to packet queue '{self.hitl_packet_id}' successfully"
            )
            return

        # Route to regular queue
        logger.info(f"Routing {file_name} to regular queue {queue_name}")
        workflow_util.enqueue_manual_review(
            queue_name=queue_name,
            message=queue_result,
            organization_id=self.organization_id,
        )
        logger.info(
            f"✅ MANUAL REVIEW: File '{file_name}' sent to manual review queue '{queue_name}' successfully"
        )

    def _push_data_to_queue(
        self,
        file_name: str,
        workflow: dict[str, Any],
        input_file_path: str,
        file_execution_id: str,
        tool_execution_result: str = None,
        api_client: Optional["InternalAPIClient"] = None,
        hitl_reason: str | None = None,
    ) -> None:
        """Handle manual review queue processing (following production pattern).

        This method replicates the backend DestinationConnector._push_to_queue logic.
        """
        if not has_manual_review_plugin():
            logger.warning(f"No manual review service available to enqueue {file_name}")
            return
        logger.info(f"Pushing {file_name} to manual review queue")
        log_file_info(
            self.workflow_log,
            file_execution_id,
            f"🔄 File '{file_name}' sending to manual review queue",
        )

        try:
            # Ensure manual review service is available and use it
            manual_review_service = self._ensure_manual_review_service(api_client)

            # Get tool execution result if not provided
            if not tool_execution_result:
                tool_execution_result = (
                    self.get_tool_execution_result_from_execution_context(
                        workflow_id=self.workflow_id,
                        execution_id=self.execution_id,
                        file_execution_id=file_execution_id,
                        organization_id=self.organization_id,
                    )
                )

            if not tool_execution_result:
                logger.warning(
                    f"No tool execution result available for {file_name}, skipping queue"
                )
                return

            # Get queue name using backend pattern
            queue_name = self._get_review_queue_name()
            # Use workflow util via plugin architecture (handles OSS/Enterprise automatically)
            workflow_util = manual_review_service.get_workflow_util()

            # Read file content based on deployment type (matching backend logic)
            if self.is_api:
                # For API deployments, read from workflow execution storage (no fallback in backend)
                file_content_base64 = self._read_file_content_for_queue(
                    input_file_path, file_name
                )
                ttl_seconds = None
            else:
                # For ETL/TASK workflows, read from source connector (like backend)
                file_content_base64 = self._read_file_from_source_connector(
                    input_file_path, file_name, workflow
                )
                ttl_seconds = workflow_util.get_hitl_ttl_seconds(str(self.workflow_id))

            # Get metadata (whisper-hash, etc.)
            metadata = self.get_metadata()
            whisper_hash = metadata.get("whisper-hash") if metadata else None
            extracted_text = metadata.get("extracted_text") if metadata else None

            # Create queue result matching backend QueueResult structure
            queue_result = QueueResult(
                file=file_name,
                status=QueueResultStatus.SUCCESS,
                result=tool_execution_result,
                workflow_id=str(self.workflow_id),
                whisper_hash=whisper_hash,
                file_execution_id=file_execution_id,
                extracted_text=extracted_text,
                ttl_seconds=ttl_seconds,
                hitl_reason=hitl_reason,
                hitl_queue_name=self.hitl_queue_name,
            )

            # Only include file_content if provided (backend API will handle it)
            if file_content_base64 is not None:
                queue_result.file_content = file_content_base64

            # Route to packet queue or regular queue based on hitl_packet_id
            self._enqueue_to_packet_or_regular_queue(
                file_name=file_name,
                queue_result=queue_result.to_dict(),
                queue_name=queue_name,
                workflow_util=workflow_util,
                ttl_seconds=ttl_seconds,
            )

            # Log successful enqueue (common for both paths)
            queue_display_name = (
                self.hitl_packet_id if self.hitl_packet_id else queue_name
            )
            log_file_info(
                self.workflow_log,
                file_execution_id,
                f"✅ File '{file_name}' sent to manual review queue '{queue_display_name}'",
            )

        except Exception as e:
            logger.error(f"Failed to push {file_name} to manual review queue: {e}")
            raise

    def _get_review_queue_name(self) -> str:
        """Generate review queue name with optional HITL override for manual review processing.

        This method replicates the backend DestinationConnector._get_review_queue_name logic.
        """
        logger.debug(f"Queue naming - hitl_queue_name={self.hitl_queue_name}")

        # Base queue format: review_queue_{org}_{workflow_id}
        base_queue_name = f"review_queue_{self.organization_id}_{str(self.workflow_id)}"

        if self.hitl_queue_name:
            # Custom HITL queue with user-specified name
            q_name = f"{base_queue_name}:{self.hitl_queue_name}"
            logger.debug(f"Using custom HITL queue: {q_name}")
        else:
            # Standard queue format for workflow-based processing
            q_name = base_queue_name
            logger.debug(f"Using standard queue name: {q_name}")

        return q_name

    def _read_file_content_for_queue(self, input_file_path: str, file_name: str) -> str:
        """Read and encode file content for queue message from execution storage.

        This method replicates the backend DestinationConnector._read_file_content_for_queue logic.
        """
        try:
            file_system = FileSystem(FileStorageType.WORKFLOW_EXECUTION)
            file_storage = file_system.get_file_storage()

            if not file_storage.exists(input_file_path):
                raise FileNotFoundError(f"File not found: {input_file_path}")

            file_bytes = file_storage.read(input_file_path, mode="rb")
            if isinstance(file_bytes, str):
                file_bytes = file_bytes.encode("utf-8")
            return base64.b64encode(file_bytes).decode("utf-8")
        except Exception as e:
            logger.error(f"Failed to read file content for {file_name}: {e}")
            raise OSError(f"Failed to read file content for queue: {e}")

    def _read_file_from_source_connector(
        self, input_file_path: str, file_name: str, workflow: dict[str, Any]
    ) -> str:
        """Read and encode file content from source connector for ETL/TASK workflows.

        This method replicates the backend logic: source_fs.open(input_file_path, "rb")
        """
        try:
            # Use source connector configuration (not destination connector!)
            if not self.source_connector_id or not self.source_connector_settings:
                # Try to get source connector info from workflow data as fallback
                source_connector_id = (
                    workflow.get("source_connector_id") if workflow else None
                )
                source_connector_settings = (
                    workflow.get("source_connector_settings") if workflow else None
                )

                if not source_connector_id or not source_connector_settings:
                    raise ValueError(
                        f"Source connector configuration not available for {file_name}"
                    )
            else:
                source_connector_id = self.source_connector_id
                source_connector_settings = self.source_connector_settings

            logger.debug(
                f"Using source connector {source_connector_id} to read {file_name}"
            )

            # Import connector operations

            # Get the source connector instance (not destination!)
            connectorkit = Connectorkit()
            connector_class = connectorkit.get_connector_class_by_connector_id(
                source_connector_id
            )
            connector_instance = connector_class(source_connector_settings)

            # Get fsspec filesystem (like backend: self.get_fsspec())
            source_fs = connector_instance.get_fsspec_fs()

            # Read file content (like backend: source_fs.open(input_file_path, "rb"))
            with source_fs.open(input_file_path, "rb") as remote_file:
                file_content = remote_file.read()
                file_content_base64 = base64.b64encode(file_content).decode("utf-8")

            logger.info(
                f"Successfully read {len(file_content)} bytes from source connector for {file_name}"
            )
            return file_content_base64

        except Exception as e:
            logger.error(
                f"Failed to read file from source connector for {file_name}: {e}"
            )
            raise OSError(f"Failed to read file from source connector: {e}")


# Alias for backward compatibility
DestinationConnector = WorkerDestinationConnector