"""
Workflow Execution Engine for EMDS

Handles workflow step execution with background threading, progress updates,
and results management.

Author: EMDS QGIS Port POC
Date: 2025-11-07
"""

import uuid
import sqlite3
from typing import Optional, Dict, Any, List, Callable
from datetime import datetime
from .qt_compat import QThread, pyqtSignal, QObject
import pandas as pd

from .bridge_loader import NetWeaverBridge, CDPBridge, PortfolioBridge
from .layer_kind import LayerKindMismatchError, LayerKindResolver
from .models import LayerKind, WorkflowStep, EngineType
from .parameter_mapper import ParameterMapper, ParameterMappingConfig, VariableMapping
from .results_manager import ResultsManager, ExecutionResult, ExecutionStatus
from . import debug_logger


class WorkflowExecutionSignals(QObject):
    """Signals for workflow execution"""
    progress_updated = pyqtSignal(str, int)  # message, percentage
    execution_started = pyqtSignal(str)  # workflow_step_id
    execution_completed = pyqtSignal(str, dict)  # execution_id, result_dict
    execution_failed = pyqtSignal(str, str)  # execution_id, error_message
    layer_created = pyqtSignal(str, str)  # layer_name, layer_path


class WorkflowExecutionThread(QThread):
    """Background thread for workflow execution"""

    def __init__(self, workflow_step: WorkflowStep, input_layer_data: Optional[pd.DataFrame],
                 engine_paths: Dict[str, str] = None, results_db_path: str = None,
                 model_blob_resolver: Optional[Callable[[Any], tuple]] = None):
        """
        Initialize execution thread

        Args:
            workflow_step: WorkflowStep to execute
            input_layer_data: Input data as DataFrame
            engine_paths: Dictionary of engine_type -> exe_path (for bridging)
            results_db_path: Path to results database
        """
        super().__init__()
        self.workflow_step = workflow_step
        self.input_layer_data = input_layer_data
        self.engine_paths = engine_paths or {}
        self.results_db_path = results_db_path
        self.execution_id = str(uuid.uuid4())
        self.signals = WorkflowExecutionSignals()
        self.is_running = True
        self.result_dataframe: Optional[pd.DataFrame] = None
        # Provenance: run from the DB-embedded model snapshot bytes when the
        # resolver supplies them (file path is the fallback). Resolver returns
        # (model_blob: Optional[bytes], source_format: str).
        self.model_blob_resolver = model_blob_resolver
        # Provenance: engine metadata captured from the bridge after execution.
        self.engine_version: str = ""
        self.engine_model_schema_version: str = ""
        self.engine_model_info: Optional[Dict[str, Any]] = None

    def emit_progress(self, message: str, percentage: int) -> None:
        """Emit progress signal"""
        self.signals.progress_updated.emit(message, percentage)

    def execute_sync(
        self,
        progress_callback: Optional[Callable[[str, int], None]] = None,
    ) -> dict:
        """Execute the workflow step synchronously on the current thread."""

        def publish(message: str, percentage: int) -> None:
            if callable(progress_callback):
                progress_callback(message, percentage)

        publish("Starting execution...", 0)

        start_time = datetime.now().isoformat()
        results_manager = ResultsManager(self.results_db_path)
        result = ExecutionResult(
            execution_id=self.execution_id,
            workflow_step_id=self.workflow_step.id,
            status=ExecutionStatus.RUNNING,
            start_time=start_time
        )

        try:
            if self.input_layer_data is None:
                publish("Loading input data...", 5)
                self.input_layer_data = self._load_input_layer_data()

            debug_logger.debug("Workflow execution started", {
                "workflow_step_id": self.workflow_step.id,
                "workflow_name": self.workflow_step.name,
                "engine_type": self.workflow_step.engine_type.value,
                "input_rows": len(self.input_layer_data.index),
                "input_columns": list(self.input_layer_data.columns),
            })

            parameters = self.workflow_step.parameters or {}
            engine_input = self.input_layer_data

            publish("Validating parameters...", 10)
            mapping_config = self._create_mapping_config()
            mapper = ParameterMapper(mapping_config)

            if not mapper.validate_mappings(self.input_layer_data.columns.tolist()):
                raise ValueError("Parameter validation failed: " + "\n".join(mapper.get_errors()))

            if self.workflow_step.engine_type == EngineType.NETWEAVER:
                debug_logger.debug("Using raw input for NetWeaver execution", {
                    "workflow_step_id": self.workflow_step.id,
                    "rows": len(engine_input.index),
                    "columns": list(engine_input.columns),
                    "field_mappings": len(parameters.get("field_mappings", [])),
                })
            elif self.workflow_step.engine_type == EngineType.CDP:
                debug_logger.debug("Using raw input for CDP mapped execution", {
                    "workflow_step_id": self.workflow_step.id,
                    "rows": len(engine_input.index),
                    "columns": list(engine_input.columns),
                    "field_mappings": len(parameters.get("field_mappings", [])),
                    "display_field": parameters.get("display_field", ""),
                    "primary_key": parameters.get("primary_key", ""),
                })
            elif self.workflow_step.engine_type == EngineType.TEMPLATE_2:
                debug_logger.debug("Using raw input for Portfolio execution", {
                    "workflow_step_id": self.workflow_step.id,
                    "rows": len(engine_input.index),
                    "columns": list(engine_input.columns),
                    "unique_field": parameters.get("unique_field", ""),
                    "cost_field": parameters.get("cost_field", ""),
                    "benefit_field": parameters.get("benefit_field", ""),
                    "strategy": parameters.get("strategy", ""),
                    "direction": parameters.get("direction", ""),
                    "overall_budget": parameters.get("overall_budget", 0.0),
                    "funding_category_field": parameters.get("funding_category_field", ""),
                })
            else:
                publish("Mapping parameters...", 20)
                mapped_data, success = mapper.map_dataframe(self.input_layer_data)

                if not success:
                    raise ValueError("Parameter mapping failed: " + "\n".join(mapper.get_errors()))

                engine_input = mapped_data
                debug_logger.debug("Mapped engine input", {
                    "workflow_step_id": self.workflow_step.id,
                    "mapped_rows": len(engine_input.index),
                    "mapped_columns": list(engine_input.columns),
                })

            publish(f"Executing with {self.workflow_step.engine_type.value}...", 40)
            results_df = self._execute_with_engine(
                engine_input,
                outer_progress_callback=publish,
            )

            if results_df is None:
                raise RuntimeError("Engine execution failed")

            self.result_dataframe = results_df

            publish("Storing results...", 80)
            results_manager.save_results_dataframe(self.execution_id, results_df)

            result.status = ExecutionStatus.COMPLETED
            result.end_time = datetime.now().isoformat()
            result.duration_ms = int((datetime.fromisoformat(result.end_time) -
                                     datetime.fromisoformat(start_time)).total_seconds() * 1000)
            result.rows_processed = len(results_df)

            results_manager.save_execution(result)
            debug_logger.success("Workflow execution completed", {
                "workflow_step_id": self.workflow_step.id,
                "execution_id": self.execution_id,
                "rows_processed": len(results_df.index),
            })

            publish("Execution completed successfully", 100)
            execution_payload = result.to_dict()
            # Surface provenance engine metadata to the persistence layer.
            execution_payload["engine_version"] = self.engine_version
            execution_payload["model_schema_version"] = self.engine_model_schema_version
            execution_payload["engine_model_info"] = self.engine_model_info
            return execution_payload

        except Exception as e:
            result.status = ExecutionStatus.FAILED
            result.end_time = datetime.now().isoformat()
            result.duration_ms = int((datetime.fromisoformat(result.end_time) -
                                     datetime.fromisoformat(start_time)).total_seconds() * 1000)
            result.error_message = str(e)
            results_manager.save_execution(result)
            debug_logger.error("Workflow execution failed", {
                "workflow_step_id": self.workflow_step.id,
                "execution_id": self.execution_id,
                "exception": str(e),
            })
            raise

    def run(self) -> None:
        """Execute workflow step (runs in background thread)"""
        self.signals.execution_started.emit(self.workflow_step.id)
        try:
            result = self.execute_sync(progress_callback=self.emit_progress)
            self.signals.execution_completed.emit(self.execution_id, result)
        except Exception as e:
            self.signals.execution_failed.emit(self.execution_id, str(e))

    def _create_mapping_config(self) -> ParameterMappingConfig:
        """Create parameter mapping configuration from workflow step"""
        config = ParameterMappingConfig(
            workflow_step_id=self.workflow_step.id,
            model_name=self.workflow_step.name,
            description=f"Mapping for {self.workflow_step.name}"
        )

        mapping_items = (self.workflow_step.parameters or {}).get("field_mappings", [])
        for mapping in mapping_items:
            source_column = mapping.get("source_column", "").strip()
            target_variable = mapping.get("target_variable", "").strip()
            if source_column and target_variable:
                config.mappings.append(
                    VariableMapping(
                        source_column=source_column,
                        target_variable=target_variable,
                    )
                )

        return config

    def _resolve_model_blob(self) -> tuple:
        """Resolve (model_blob, source_format) for run-from-snapshot, or (None, "")."""
        resolver = getattr(self, "model_blob_resolver", None)
        if resolver is None:
            return None, ""
        try:
            blob, source_format = resolver(self.workflow_step)
            return blob, (source_format or "")
        except Exception:
            return None, ""

    def _capture_engine_provenance(self, bridge) -> None:
        """Pull provenance metadata off the bridge after an engine call."""
        try:
            self.engine_version = getattr(bridge, "last_engine_version", "") or ""
            self.engine_model_schema_version = getattr(bridge, "last_model_schema_version", "") or ""
            self.engine_model_info = getattr(bridge, "last_model_info", None)
        except Exception:
            pass

    @staticmethod
    def _quote_identifier(identifier: str) -> str:
        return '"' + str(identifier).replace('"', '""') + '"'

    @staticmethod
    def _preferred_stable_key_column(columns: List[str]) -> str:
        """Pick the best stable feature identity column carried through overlays."""
        by_lower = {str(column).strip().lower(): str(column) for column in columns}
        for candidate in (
            "input_objectid",
            "objectid",
            "source_objectid",
            "emds_source_feature_id",
            "input_fid_attr",
            "ogc_fid",
            "fid",
        ):
            if candidate in by_lower:
                return by_lower[candidate]
        return ""

    def _sort_dataframe_by_stable_key(self, df: pd.DataFrame) -> pd.DataFrame:
        key_column = self._preferred_stable_key_column([str(column) for column in df.columns])
        if not key_column or key_column not in df.columns:
            return df
        try:
            sorted_df = df.copy()
            sorted_df["_emds_sort_key"] = pd.to_numeric(sorted_df[key_column], errors="coerce")
            sorted_df = sorted_df.sort_values(
                by=["_emds_sort_key", key_column],
                kind="mergesort",
                na_position="last",
            ).drop(columns=["_emds_sort_key"])
            sorted_df = sorted_df.reset_index(drop=True)
            debug_logger.debug("Sorted execution input by stable key", {
                "workflow_step_id": self.workflow_step.id,
                "key_column": key_column,
                "rows": len(sorted_df.index),
            })
            return sorted_df
        except Exception as exc:
            debug_logger.warning("Could not sort execution input by stable key", {
                "workflow_step_id": self.workflow_step.id,
                "key_column": key_column,
                "exception": str(exc),
            })
            return df

    def _load_input_layer_data(self) -> pd.DataFrame:
        """Load real input data from the live QGIS layer or project GeoPackage."""
        parameters = self.workflow_step.parameters or {}
        input_layer_name = parameters.get("input_layer_name", "")
        input_table_name = parameters.get("input_table_name", "")
        database_path = parameters.get("database_path", "")

        debug_logger.debug("Loading input data for execution", {
            "workflow_step_id": self.workflow_step.id,
            "input_layer_name": input_layer_name,
            "input_table_name": input_table_name,
            "database_path": database_path,
        })

        if database_path and input_table_name:
            try:
                conn = sqlite3.connect(database_path)
                columns = [
                    str(row[1])
                    for row in conn.execute(
                        f"PRAGMA table_info({self._quote_identifier(input_table_name)})"
                    ).fetchall()
                ]
                key_column = self._preferred_stable_key_column(columns)
                order_clause = f" ORDER BY {self._quote_identifier(key_column)}" if key_column else ""
                df = pd.read_sql_query(
                    f"SELECT * FROM {self._quote_identifier(input_table_name)}{order_clause}",
                    conn,
                )
                conn.close()
                debug_logger.debug("Loaded input data from GeoPackage table", {
                    "workflow_step_id": self.workflow_step.id,
                    "database_path": database_path,
                    "input_table_name": input_table_name,
                    "order_by": key_column,
                    "rows": len(df.index),
                    "columns": list(df.columns),
                })
                return df
            except Exception as exc:
                debug_logger.error("Failed to load input data from GeoPackage", {
                    "exception": str(exc),
                    "database_path": database_path,
                    "input_table_name": input_table_name,
                })

        try:
            from qgis.core import QgsProject

            for layer in QgsProject.instance().mapLayers().values():
                if layer.name().lower() == input_layer_name.lower():
                    # Raster-support Phase 2 guard: today's engines
                    # (NetWeaver / CDP / Portfolio) all consume a
                    # vector-attribute DataFrame; calling layer.fields()
                    # / layer.getFeatures() on a raster would raise an
                    # AttributeError deep in the loop. Fail fast with
                    # a typed error so the run-task UI can surface a
                    # clean message instead of a stack trace.
                    LayerKindResolver.require(
                        layer,
                        LayerKind.VECTOR,
                        layer_name=input_layer_name,
                    )
                    rows = []
                    field_names = [field.name() for field in layer.fields()]
                    for feature in layer.getFeatures():
                        row = {field_name: feature[field_name] for field_name in field_names}
                        rows.append(row)
                    df = self._sort_dataframe_by_stable_key(pd.DataFrame(rows))
                    debug_logger.debug("Loaded input data from live QGIS layer", {
                        "workflow_step_id": self.workflow_step.id,
                        "input_layer_name": input_layer_name,
                        "rows": len(df.index),
                        "columns": list(df.columns),
                    })
                    return df
        except LayerKindMismatchError:
            # Re-raise — this is a real configuration error, not a
            # "QGIS layer load failed" fallback case. The wrapping
            # try/except below would swallow it as a generic warning.
            raise
        except Exception as exc:
            debug_logger.warning("Falling back from QGIS layer load", {"exception": str(exc)})

        raise ValueError("Unable to load input data for workflow execution")

    def _validate_netweaver_input_mappings(
        self,
        input_data: pd.DataFrame,
        parameters: Dict[str, Any],
    ) -> None:
        """Fail early when the wizard mappings do not match the actual execution table."""
        mapping_items = parameters.get("field_mappings", []) or []
        if not mapping_items:
            return

        available_columns = {str(column).strip().lower(): str(column) for column in input_data.columns}
        missing_columns: List[str] = []
        resolved_columns: List[str] = []

        for mapping in mapping_items:
            source_column = str(mapping.get("source_column", "")).strip()
            if not source_column:
                continue
            resolved = available_columns.get(source_column.lower())
            if resolved:
                resolved_columns.append(f"{source_column}->{resolved}")
            else:
                missing_columns.append(source_column)

        if missing_columns:
            table_name = parameters.get("input_table_name", "")
            raise ValueError(
                "The NetWeaver field mappings do not match the execution input table. "
                f"Missing columns: {', '.join(missing_columns)}. "
                f"Input table: {table_name or '(unknown)'}. "
                f"Available columns: {', '.join(map(str, input_data.columns))}"
            )

        debug_logger.debug("Validated NetWeaver field mappings against input data", {
            "workflow_step_id": self.workflow_step.id,
            "resolved_mappings": resolved_columns,
            "column_count": len(input_data.columns),
        })

    def _execute_with_engine(
        self,
        input_data: pd.DataFrame,
        outer_progress_callback: Optional[Callable[[str, int], None]] = None,
    ) -> Optional[pd.DataFrame]:
        """
        Execute workflow step with appropriate engine

        Args:
            input_data: Mapped input data

        Returns:
            Results DataFrame or None
        """
        engine_type = self.workflow_step.engine_type
        parameters = self.workflow_step.parameters or {}

        try:
            if engine_type == EngineType.NETWEAVER:
                return self._execute_netweaver(input_data, parameters, outer_progress_callback=outer_progress_callback)
            elif engine_type == EngineType.CDP:
                return self._execute_cdp(input_data, parameters, outer_progress_callback=outer_progress_callback)
            elif engine_type == EngineType.TEMPLATE_2:
                return self._execute_portfolio(input_data, parameters, outer_progress_callback=outer_progress_callback)
            else:
                raise ValueError(f"Unsupported engine type: {engine_type}")

        except Exception as e:
            raise RuntimeError(f"Engine execution failed: {str(e)}")

    def _execute_netweaver(self, input_data: pd.DataFrame,
                          parameters: Dict[str, Any],
                          outer_progress_callback: Optional[Callable[[str, int], None]] = None) -> Optional[pd.DataFrame]:
        """Execute with NetWeaver engine"""
        netweaver_path = self.engine_paths.get('netweaver')
        if not netweaver_path:
            raise ValueError("NetWeaver engine path not configured")

        try:
            bridge = NetWeaverBridge(netweaver_path)
            model_path = parameters.get('model_file')

            if not model_path:
                raise ValueError("Model file path not specified in parameters")

            debug_logger.debug("Calling NetWeaver bridge", {
                "workflow_step_id": self.workflow_step.id,
                "engine_path": netweaver_path,
                "model_path": model_path,
                "rows": len(input_data.index),
                "columns": list(input_data.columns),
                "selected_topics": parameters.get("selected_topics", []),
                "map_topics": parameters.get("map_topics", []),
                "field_mapping_count": len(parameters.get("field_mappings", [])),
            })

            self._validate_netweaver_input_mappings(input_data, parameters)

            # Define progress callback to update UI at coarse milestones so the
            # dialog stays readable on large runs.
            last_reported_bucket = {"value": -1}

            def progress_update(current: int, total: int, percentage: int):
                """Progress callback from engine"""
                bucket = min(100, max(0, (percentage // 5) * 5))
                if bucket == last_reported_bucket["value"] and percentage < 100:
                    return
                last_reported_bucket["value"] = bucket
                message = f"Processing NetWeaver model... {bucket}% complete"
                # Map engine progress (0-100%) to our progress range (40-80%)
                ui_percentage = 40 + int(bucket * 0.4)
                if callable(outer_progress_callback):
                    outer_progress_callback(message, ui_percentage)
                else:
                    self.emit_progress(message, ui_percentage)

            # Process data through NetWeaver with progress monitoring. When a
            # model snapshot is available, run from its embedded bytes so the
            # exact stored model executes (file path is the fallback).
            model_blob, model_source_format = self._resolve_model_blob()
            results_df = bridge.process_data(
                data=input_data,
                model_path=model_path,
                parameters={
                    "selected_topics": parameters.get("selected_topics", []),
                    "map_topics": parameters.get("map_topics", []),
                    "field_mappings": parameters.get("field_mappings", []),
                    "workflow_step_id": self.workflow_step.id,
                    "workflow_step_name": self.workflow_step.name,
                },
                use_arrow=True,
                progress_callback=progress_update,
                model_blob=model_blob,
                model_source_format=model_source_format,
            )

            self._capture_engine_provenance(bridge)
            return results_df

        except Exception as e:
            raise RuntimeError(f"NetWeaver execution failed: {str(e)}")

    def _execute_cdp(self, input_data: pd.DataFrame,
                    parameters: Dict[str, Any],
                    outer_progress_callback: Optional[Callable[[str, int], None]] = None) -> Optional[pd.DataFrame]:
        """Execute with CDP engine"""
        cdp_path = self.engine_paths.get('cdp')
        if not cdp_path:
            raise ValueError("CDP engine path not configured")

        try:
            bridge = CDPBridge(cdp_path)
            model_path = parameters.get('model_file')

            if not model_path:
                raise ValueError("Model file path not specified in parameters")

            # Define progress callback to update UI
            def progress_update(current: int, total: int, percentage: int):
                """Progress callback from engine"""
                message = f"CDP processing row {current:,} of {total:,}"
                # Map engine progress (0-100%) to our progress range (40-80%)
                ui_percentage = 40 + int(percentage * 0.4)
                if callable(outer_progress_callback):
                    outer_progress_callback(message, ui_percentage)
                else:
                    self.emit_progress(message, ui_percentage)

            # Process data through CDP with progress monitoring. Run from the
            # embedded snapshot bytes when available (file path is the fallback).
            model_blob, model_source_format = self._resolve_model_blob()
            results_df = bridge.process_with_mapping(
                data=input_data,
                model_path=model_path,
                field_mappings=parameters.get("field_mappings", []),
                mapping_rows=parameters.get("mapping_rows", []),
                display_field=parameters.get("display_field", ""),
                primary_key=parameters.get("primary_key", ""),
                input_table_name=parameters.get("input_table_name", ""),
                validation_settings=parameters.get("validation_settings", {}),
                progress_callback=progress_update,
                model_blob=model_blob,
                model_source_format=model_source_format,
            )

            self._capture_engine_provenance(bridge)
            return results_df

        except Exception as e:
            raise RuntimeError(f"CDP execution failed: {str(e)}")

    def _execute_portfolio(
        self,
        input_data: pd.DataFrame,
        parameters: Dict[str, Any],
        outer_progress_callback: Optional[Callable[[str, int], None]] = None,
    ) -> Optional[pd.DataFrame]:
        """
        Execute a Portfolio workflow step using the pure-Python PortfolioBridge.

        No external engine executable is required: the greedy knapsack
        allocation runs in-process. We still funnel progress through the
        standard callback plumbing so the dialog stays responsive.
        """
        try:
            bridge = PortfolioBridge()

            required_field_names = [
                parameters.get("unique_field", ""),
                parameters.get("cost_field", ""),
                parameters.get("benefit_field", ""),
            ]
            missing = [name for name in required_field_names if not str(name).strip()]
            if missing:
                raise ValueError(
                    "Portfolio parameters are missing required field selections "
                    "(unique/cost/benefit)."
                )

            available_columns = {
                str(column).strip().lower(): str(column)
                for column in input_data.columns
            }
            for role, name in (
                ("unique_field", parameters.get("unique_field", "")),
                ("cost_field", parameters.get("cost_field", "")),
                ("benefit_field", parameters.get("benefit_field", "")),
            ):
                if name and name.strip().lower() not in available_columns:
                    raise ValueError(
                        f"Portfolio {role} '{name}' is not present in the input data. "
                        f"Available columns: {', '.join(map(str, input_data.columns))}"
                    )

            debug_logger.debug("Calling Portfolio bridge", {
                "workflow_step_id": self.workflow_step.id,
                "rows": len(input_data.index),
                "columns": list(input_data.columns),
                "unique_field": parameters.get("unique_field", ""),
                "cost_field": parameters.get("cost_field", ""),
                "benefit_field": parameters.get("benefit_field", ""),
                "strategy": parameters.get("strategy", ""),
                "direction": parameters.get("direction", ""),
                "overall_budget": parameters.get("overall_budget", 0.0),
                "funding_category_field": parameters.get("funding_category_field", ""),
                "silo_budgets": parameters.get("silo_budgets"),
                "funding_percent_rows": len(parameters.get("funding_percents", []) or []),
            })

            last_reported_bucket = {"value": -1}

            def progress_update(current: int, total: int, percentage: int):
                """Progress callback from the portfolio engine."""
                bucket = min(100, max(0, (percentage // 5) * 5))
                if bucket == last_reported_bucket["value"] and percentage < 100:
                    return
                last_reported_bucket["value"] = bucket
                message = (
                    f"Allocating portfolio... {current:,} of {total:,} "
                    f"candidates ({bucket}%)"
                )
                # Map engine progress (0-100%) into the outer UI range (40-80%)
                ui_percentage = 40 + int(bucket * 0.4)
                if callable(outer_progress_callback):
                    outer_progress_callback(message, ui_percentage)
                else:
                    self.emit_progress(message, ui_percentage)

            results_df = bridge.process_data(
                data=input_data,
                parameters=parameters,
                progress_callback=progress_update,
            )

            report = getattr(bridge, "last_report", None)
            if report is not None:
                debug_logger.debug("Portfolio run report", {
                    "workflow_step_id": self.workflow_step.id,
                    "selected_count": report.selected_count,
                    "rows_considered": report.rows_considered,
                    "total_spent": report.total_spent,
                    "total_budget": report.total_budget,
                    "budget_left": report.budget_left,
                    "efficacy": report.efficacy,
                    "processing_time_ms": report.processing_time_ms,
                })
                # Stash the report onto the workflow step so the dock widget
                # can surface it during persistence / tree display.
                metadata = dict(self.workflow_step.metadata or {})
                metadata["portfolio_report"] = {
                    "total_spent": report.total_spent,
                    "total_budget": report.total_budget,
                    "budget_left": report.budget_left,
                    "left_silo_a": report.left_silo_a,
                    "left_silo_b": report.left_silo_b,
                    "left_silo_c": report.left_silo_c,
                    "selected_count": report.selected_count,
                    "efficacy": report.efficacy,
                    "rows_considered": report.rows_considered,
                    "processing_time_ms": report.processing_time_ms,
                }
                self.workflow_step.metadata = metadata

            return results_df

        except Exception as e:
            raise RuntimeError(f"Portfolio execution failed: {str(e)}")


class WorkflowExecutor(QObject):
    """Manages workflow execution and results"""

    execution_started = pyqtSignal(str)  # execution_id
    execution_completed = pyqtSignal(str, dict)  # execution_id, result
    execution_failed = pyqtSignal(str, str)  # execution_id, error_message

    def __init__(self, engine_paths: Dict[str, str] = None,
                 results_db_path: str = None):
        """
        Initialize workflow executor

        Args:
            engine_paths: Dictionary of engine_type -> exe_path
            results_db_path: Path to results database
        """
        super().__init__()
        self.engine_paths = engine_paths or {}
        self.results_db_path = results_db_path
        self.execution_threads: Dict[str, WorkflowExecutionThread] = {}
        self.results_cache: Dict[str, pd.DataFrame] = {}
        self.results_manager = ResultsManager(results_db_path)
        # Provenance: optional resolver(workflow_step) -> (model_blob, source_format)
        # so live runs execute the DB-embedded snapshot bytes. Set by the dock.
        self.model_blob_resolver: Optional[Callable[[Any], tuple]] = None

    def execute_workflow_step(self, workflow_step: WorkflowStep,
                             input_layer_data: pd.DataFrame) -> str:
        """
        Execute a workflow step asynchronously

        Args:
            workflow_step: WorkflowStep to execute
            input_layer_data: Input data as DataFrame

        Returns:
            Execution ID
        """
        # Create execution thread
        thread = WorkflowExecutionThread(
            workflow_step=workflow_step,
            input_layer_data=input_layer_data,
            engine_paths=self.engine_paths,
            results_db_path=self.results_db_path,
            model_blob_resolver=self.model_blob_resolver,
        )

        # Connect signals
        thread.signals.execution_completed.connect(
            lambda exec_id, result: self._on_execution_completed(exec_id, result)
        )
        thread.signals.execution_failed.connect(
            lambda exec_id, error: self._on_execution_failed(exec_id, error)
        )

        # Store and start
        execution_id = thread.execution_id
        self.execution_threads[execution_id] = thread
        self.execution_started.emit(execution_id)
        thread.start()

        return execution_id

    def execute_workflow_step_sync(
        self,
        workflow_step: WorkflowStep,
        input_layer_data: Optional[pd.DataFrame] = None,
        progress_callback: Optional[Callable[[str, int], None]] = None,
        *,
        results_db_path: Optional[str] = None,
    ) -> tuple[str, dict, Optional[pd.DataFrame]]:
        """Execute a workflow step synchronously and return execution artifacts.

        Pass ``results_db_path`` to persist into a session staging database
        (Scenario Builder Test) instead of the canonical ~/.emds/results.db.
        """

        thread = WorkflowExecutionThread(
            workflow_step=workflow_step,
            input_layer_data=input_layer_data,
            engine_paths=self.engine_paths,
            results_db_path=results_db_path if results_db_path is not None else self.results_db_path,
            model_blob_resolver=self.model_blob_resolver,
        )
        result = thread.execute_sync(progress_callback=progress_callback)
        execution_id = thread.execution_id
        if thread.result_dataframe is not None:
            self.results_cache[execution_id] = thread.result_dataframe
        return execution_id, result, thread.result_dataframe

    def get_execution_progress_signals(self, execution_id: str) -> Optional[WorkflowExecutionSignals]:
        """
        Get progress signals for an execution

        Args:
            execution_id: Execution ID

        Returns:
            WorkflowExecutionSignals or None
        """
        thread = self.execution_threads.get(execution_id)
        return thread.signals if thread else None

    def get_results(self, execution_id: str) -> Optional[pd.DataFrame]:
        """
        Get results for an execution

        Args:
            execution_id: Execution ID

        Returns:
            Results DataFrame or None
        """
        if execution_id in self.results_cache:
            return self.results_cache[execution_id]
        return self.results_manager.get_execution_results_dataframe(execution_id)

    def get_execution_history(self, workflow_step_id: str = None) -> list:
        """
        Get execution history

        Args:
            workflow_step_id: Filter by workflow step (optional)

        Returns:
            List of ExecutionResult objects
        """
        if workflow_step_id:
            return self.results_manager.get_workflow_executions(workflow_step_id)
        else:
            return self.results_manager.get_execution_history()

    def _on_execution_completed(self, execution_id: str, result: dict) -> None:
        """Handle execution completion"""
        thread = self.execution_threads.get(execution_id)
        if thread and thread.result_dataframe is not None:
            self.results_cache[execution_id] = thread.result_dataframe
        self.execution_completed.emit(execution_id, result)

        # Clean up thread
        if execution_id in self.execution_threads:
            del self.execution_threads[execution_id]

    def _on_execution_failed(self, execution_id: str, error_message: str) -> None:
        """Handle execution failure"""
        self.execution_failed.emit(execution_id, error_message)
        self.results_cache.pop(execution_id, None)

        # Clean up thread
        if execution_id in self.execution_threads:
            del self.execution_threads[execution_id]

    def cancel_execution(self, execution_id: str) -> bool:
        """
        Cancel a running execution

        Args:
            execution_id: Execution ID

        Returns:
            True if cancelled successfully
        """
        thread = self.execution_threads.get(execution_id)
        if thread and thread.is_running:
            thread.is_running = False
            thread.quit()
            thread.wait()
            return True
        return False
