Merge pull request #253 from webtech-network/234-td-separate-pipelineexecution-summary-logic

Author: ArthurCRodrigues

autograder/models/dataclass/step_result.py

@@ -30,6 +30,7 @@ class StepResult(Generic[T]):
     data: T
     status: StepStatus = StepStatus.SUCCESS
     error: Optional[str] = None
+    error_data: Any = None
     original_input: Any = None
 
     @property
@@ -43,6 +44,6 @@ def success(cls, step: StepName, data: T) -> "StepResult[T]":
         return cls(step=step, data=data, status=StepStatus.SUCCESS)
 
     @classmethod
-    def fail(cls, step: StepName, error: str, data: Optional[T] = None) -> "StepResult[T]":
+    def fail(cls, step: StepName, error: str, data: Optional[T] = None, error_data: Any = None) -> "StepResult[T]":
         """Creates a failed StepResult."""
-        return cls(step=step, data=data, status=StepStatus.FAIL, error=error)
+        return cls(step=step, data=data, status=StepStatus.FAIL, error=error, error_data=error_data)

autograder/models/pipeline_execution.py

@@ -44,151 +44,7 @@ def add_step_result(self, step_result: StepResult) -> 'PipelineExecution':
         self.step_results.append(step_result)
         return self
 
-    def get_pipeline_execution_summary(self) -> Dict[str, Any]:
-        """
-        Generate a detailed summary of the pipeline execution for API responses.
-
-        Returns:
-            Dictionary containing execution status, steps, and error details
-        """
-        execution_time_ms = int((time.time() - self.start_time) * 1000) if self.start_time else 0
-
-        # Determine failed step if any
-        failed_step = None
-        for step in self.step_results:
-            if step.status == StepStatus.FAIL:
-                failed_step = step.step.value
-                break
-
-        # Build step details
-        steps = []
-        for step in self.step_results:
-            if step.step == StepName.BOOTSTRAP:
-                continue  # Skip bootstrap step in output
-
-            step_info = {
-                "name": step.step.value,
-                "status": step.status.value,
-            }
-
-            # Add execution time if available
-            if hasattr(step, 'execution_time_ms'):
-                step_info["execution_time_ms"] = step.execution_time_ms
-
-            # Add success message for completed steps
-            if step.status == StepStatus.SUCCESS:
-                if step.step == StepName.LOAD_TEMPLATE:
-                    step_info["message"] = "Template loaded successfully"
-                elif step.step == StepName.BUILD_TREE:
-                    step_info["message"] = "Criteria tree built successfully"
-                elif step.step == StepName.PRE_FLIGHT:
-                    step_info["message"] = "All preflight checks passed"
-                elif step.step == StepName.GRADE:
-                    if self.result and self.result.final_score is not None:
-                        step_info["message"] = f"Grading completed: {self.result.final_score}/100"
-                    else:
-                        step_info["message"] = "Grading completed"
-                elif step.step == StepName.FEEDBACK:
-                    step_info["message"] = "Feedback generated"
-                elif step.step == StepName.EXPORTER:
-                    step_info["message"] = "Results exported"
-
-            # Add error details for failed steps
-            elif step.status == StepStatus.FAIL and step.error:
-                step_info["message"] = step.error.split('\n')[0]  # First line of error
-                step_info["error_details"] = self._extract_error_details(step)
-
-            steps.append(step_info)
-
-        # Count planned vs completed steps
-        total_planned = 7  # BOOTSTRAP, LOAD_TEMPLATE, BUILD_TREE, PRE_FLIGHT, GRADE, FEEDBACK, EXPORT
-        completed = len([s for s in self.step_results if s.step != StepName.BOOTSTRAP])
-
-        return {
-            "status": self.status.value if self.status != PipelineStatus.EMPTY else "unknown",
-            "failed_at_step": failed_step,
-            "total_steps_planned": total_planned,
-            "steps_completed": completed,
-            "execution_time_ms": execution_time_ms,
-            "steps": steps
-        }
-
-    def _extract_error_details(self, step: StepResult) -> Dict[str, Any]:
-        """Extract structured error details from a failed step."""
-        error_details = {}
-
-        if step.step == StepName.PRE_FLIGHT:
-            # Parse preflight errors
-            error_text = step.error
-
-            # Check if it's a missing file error
-            if "Arquivo ou diretório obrigatório não encontrado" in error_text or "Required file" in error_text:
-                error_details["error_type"] = "required_file_missing"
-                error_details["phase"] = "required_files"
-                # Extract file name if possible
-                if "`'" in error_text:
-                    start = error_text.find("`'") + 2
-                    end = error_text.find("'`", start)
-                    if end > start:
-                        error_details["missing_file"] = error_text[start:end]
-
-            # Check if it's a setup command error
-            elif "Setup command" in error_text:
-                error_details["error_type"] = "setup_command_failed"
-                error_details["phase"] = "setup_commands"
-
-                # Extract command name
-                if "Setup command '" in error_text:
-                    start = error_text.find("Setup command '") + 15
-                    end = error_text.find("' failed", start)
-                    if end > start:
-                        error_details["command_name"] = error_text[start:end]
-
-                # Extract command
-                if "**Command:** `" in error_text:
-                    start = error_text.find("**Command:** `") + 14
-                    end = error_text.find("`", start)
-                    if end > start:
-                        if "failed_command" not in error_details:
-                            error_details["failed_command"] = {}
-                        error_details["failed_command"]["command"] = error_text[start:end]
-
-                # Extract exit code
-                if "exit code " in error_text:
-                    try:
-                        start = error_text.find("exit code ") + 10
-                        end = start
-                        while end < len(error_text) and error_text[end].isdigit():
-                            end += 1
-                        if end > start:
-                            if "failed_command" not in error_details:
-                                error_details["failed_command"] = {}
-                            error_details["failed_command"]["exit_code"] = int(error_text[start:end])
-                    except (ValueError, IndexError):
-                        pass
-
-                # Extract stderr
-                if "**Error Output (stderr):**" in error_text:
-                    start = error_text.find("**Error Output (stderr):**") + 26
-                    # Find the code block
-                    start = error_text.find("```", start) + 3
-                    end = error_text.find("```", start)
-                    if end > start:
-                        if "failed_command" not in error_details:
-                            error_details["failed_command"] = {}
-                        error_details["failed_command"]["stderr"] = error_text[start:end].strip()
-
-                # Extract stdout if present
-                if "**Output (stdout):**" in error_text:
-                    start = error_text.find("**Output (stdout):**") + 20
-                    start = error_text.find("```", start) + 3
-                    end = error_text.find("```", start)
-                    if end > start:
-                        if "failed_command" not in error_details:
-                            error_details["failed_command"] = {}
-                        error_details["failed_command"]["stdout"] = error_text[start:end].strip()
 
-        return error_details
 
     def get_step_result(self, step_name: StepName) -> StepResult:
         for step_result in self.step_results:

autograder/serializers/pipeline_execution_serializer.py

@@ -0,0 +1,135 @@
+import time
+from typing import Dict, Any
+
+from autograder.models.dataclass.step_result import StepResult, StepName, StepStatus
+from autograder.models.pipeline_execution import PipelineExecution, PipelineStatus
+from autograder.models.dataclass.preflight_error import PreflightCheckType
+
+
+class PipelineExecutionSerializer:
+    """
+    Handles the serialization of PipelineExecution into dictionary payloads
+    suitable for API responses and analytical views.
+    """
+
+    @classmethod
+    def serialize(cls, execution: PipelineExecution) -> Dict[str, Any]:
+        """
+        Generate a detailed summary of the pipeline execution for API responses.
+
+        Returns:
+            Dictionary containing execution status, steps, and error details
+        """
+        execution_time_ms = int((time.time() - execution.start_time) * 1000) if execution.start_time else 0
+
+        # Determine failed step if any
+        failed_step = None
+        for step in execution.step_results:
+            if step.status == StepStatus.FAIL:
+                failed_step = step.step.value
+                break
+
+        # Build step details
+        steps = []
+        for step in execution.step_results:
+            if step.step == StepName.BOOTSTRAP:
+                continue  # Skip bootstrap step in output
+
+            step_info = {
+                "name": step.step.value,
+                "status": step.status.value,
+            }
+
+            # Add execution time if available
+            if hasattr(step, 'execution_time_ms'):
+                step_info["execution_time_ms"] = step.execution_time_ms
+
+            # Add success message for completed steps
+            if step.status == StepStatus.SUCCESS:
+                step_info["message"] = cls._get_success_message(step, execution)
+
+            # Add error details for failed steps
+            elif step.status == StepStatus.FAIL and step.error:
+                step_info["message"] = step.error.split('\n')[0]  # First line of error
+                step_info["error_details"] = cls._extract_error_details(step)
+
+            steps.append(step_info)
+
+        # Count planned vs completed steps
+        total_planned = 7  # BOOTSTRAP, LOAD_TEMPLATE, BUILD_TREE, PRE_FLIGHT, GRADE, FEEDBACK, EXPORT
+        completed = len([s for s in execution.step_results if s.step != StepName.BOOTSTRAP])
+
+        return {
+            "status": execution.status.value if execution.status != PipelineStatus.EMPTY else "unknown",
+            "failed_at_step": failed_step,
+            "total_steps_planned": total_planned,
+            "steps_completed": completed,
+            "execution_time_ms": execution_time_ms,
+            "steps": steps
+        }
+
+    @classmethod
+    def _get_success_message(cls, step: StepResult, execution: PipelineExecution) -> str:
+        """Returns success message for a specific pipeline step."""
+        if step.step == StepName.LOAD_TEMPLATE:
+            return "Template loaded successfully"
+        elif step.step == StepName.BUILD_TREE:
+            return "Criteria tree built successfully"
+        elif step.step == StepName.PRE_FLIGHT:
+            return "All preflight checks passed"
+        elif step.step == StepName.GRADE:
+            if execution.result and execution.result.final_score is not None:
+                return f"Grading completed: {execution.result.final_score}/100"
+            return "Grading completed"
+        elif step.step == StepName.FEEDBACK:
+            return "Feedback generated"
+        elif step.step == StepName.EXPORTER:
+            return "Results exported"
+        return ""
+
+    @classmethod
+    def _extract_error_details(cls, step: StepResult) -> Dict[str, Any]:
+        """
+        Extract structured error details from a failed step using its structured error_data.
+        This provides backward compatibility with the legacy text-based regex parser.
+        """
+        error_details = {}
+
+        if step.step == StepName.PRE_FLIGHT and step.error_data:
+            # error_data is a list of PreflightError objects. 
+            # We map properties from the first critical error for legacy backward compatibility.
+            if isinstance(step.error_data, list) and len(step.error_data) > 0:
+                first_error = step.error_data[0]
+                
+                # Check for Dictionary or Object access depending on how it's passed at runtime
+                err_type = first_error.type if hasattr(first_error, 'type') else first_error.get('type')
+                err_details = first_error.details if hasattr(first_error, 'details') else first_error.get('details', {})
+
+                if err_type == PreflightCheckType.FILE_CHECK:
+                    error_details["error_type"] = "required_file_missing"
+                    error_details["phase"] = "required_files"
+                    if err_details and "missing_file" in err_details:
+                        error_details["missing_file"] = err_details["missing_file"]
+                        
+                elif err_type == PreflightCheckType.SETUP_COMMAND:
+                    error_details["error_type"] = "setup_command_failed"
+                    error_details["phase"] = "setup_commands"
+                    if err_details:
+                        if "command_name" in err_details:
+                            error_details["command_name"] = err_details["command_name"]
+                            
+                        failed_command = {}
+                        if "command" in err_details:
+                            failed_command["command"] = err_details["command"]
+                        if "exit_code" in err_details:
+                            failed_command["exit_code"] = err_details["exit_code"]
+                        if "stderr" in err_details:
+                            failed_command["stderr"] = err_details["stderr"]
+                        if "stdout" in err_details:
+                            failed_command["stdout"] = err_details["stdout"]
+                            
+                        # Preserve legacy nested structure
+                        if failed_command:
+                            error_details["failed_command"] = failed_command
+
+        return error_details

autograder/steps/pre_flight_step.py

@@ -70,6 +70,7 @@ def execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
                         data=sandbox,  # sandbox is None here, which is correct
                         status=StepStatus.FAIL,
                         error=error_msg,
+                        error_data=self._pre_flight_service.fatal_errors,
                         original_input=pipeline_exec
                         ))
 
@@ -100,6 +101,7 @@ def execute(self, pipeline_exec: PipelineExecution) -> PipelineExecution:
                     data=sandbox,#Return Sandbox Here anyway? (How to deal with sandbox destruction)
                     status=StepStatus.FAIL,
                     error=error_msg,
+                    error_data=self._pre_flight_service.fatal_errors,
                     original_input=pipeline_exec
                 ))
 

docs/architecture/core_structures.md

@@ -245,30 +245,31 @@ class PipelineExecution:
     start_time: float               # Execution start timestamp
 ```
 
-### Methods
+### Pipeline Execution Summary
 
-```python
-def get_pipeline_execution_summary() -> dict:
-    """
-    Generate detailed summary for API responses.
-    
-    Returns:
+The `PipelineExecutionSerializer` (`autograder/serializers/pipeline_execution_serializer.py`) handles the conversion of an execution into a dictionary for API responses.
+
+#### `PipelineExecutionSerializer.serialize(execution) -> dict`
+
+Generates detailed summary for API responses.
+
+**Returns:**
+```json
+{
+    "status": "success" | "failed",
+    "failed_at_step": str | None,
+    "total_steps_planned": int,
+    "steps_completed": int,
+    "execution_time_ms": int,
+    "steps": [
         {
-            "status": "success" | "failed",
-            "failed_at_step": str | None,
-            "total_steps_planned": int,
-            "steps_completed": int,
-            "execution_time_ms": int,
-            "steps": [
-                {
-                    "name": "PRE_FLIGHT",
-                    "status": "success" | "fail",
-                    "message": str,
-                    "error_details": dict | None  # Present if step failed
-                }
-            ]
+            "name": "PreFlightStep",
+            "status": "success" | "fail",
+            "message": str,
+            "error_details": dict | None  # Present if step failed
         }
-    """
+    ]
+}
 ```
 
 ### Example Summary (Success)

docs/guides/github_module.md

@@ -17,7 +17,7 @@ In short, the adopted solution consists of running a Docker container with the d
 1. [Action.yml](../../action.yml)
    Defines the inputs received by the action, describes the output as a JSON evaluation, and runs the container with the received data as environment variables.
 
-2. [Dockerfile](../../github_action/Dockerfile.actions)
+2. [Dockerfile](../../Dockerfile.actions)
    Builds the container that installs the Autograder repository and sets [Entrypoint.sh](../../github_action/entrypoint.sh) as the script to be executed when the container starts.
 
 3. [Entrypoint.sh](../../github_action/entrypoint.sh)