Method _run_sequential

ci/util/running_process_group.py:629–683

Execute processes in sequence.

(self)

        return any_activity

    def _run_sequential(self) -> list[ProcessTiming]:
        """Execute processes in sequence."""
        completed_timings: list[ProcessTiming] = []

        # Enable status monitoring
        self._status_monitoring_active = True
        try:
            for process in self.processes:
                print(f"Running: {process.get_command_str()}", flush=True)

                # Track process start time
                self._track_process_start(process)

                # Start the process if not already running
                if process.proc is None:
                    process.start()

                try:
                    exit_code = cast(int, process.wait())

                    # Collect timing data
                    if process.duration is not None:
                        timing = ProcessTiming(
                            name=_get_friendly_test_name(process.command),
                            duration=process.duration,
                            command=str(process.command),
                        )
                        completed_timings.append(timing)

                    # Check for failure
                    if exit_code != 0:
                        error_snippet = extract_error_snippet(
                            _get_output_lines(process)
                        )
                        failure = TestFailureInfo(
                            test_name=_extract_test_name(process.command),
                            command=str(process.command),
                            return_code=exit_code,
                            output=error_snippet,
                            error_type="exit_failure",
                        )
                        raise TestExecutionFailedException(
                            f"Process failed with exit code {exit_code}", [failure]
                        )

                except KeyboardInterrupt as ki:
                    handle_keyboard_interrupt(ki)
                    raise
                except Exception as e:
                    print(f"Process failed: {process.get_command_str()} - {e}")
                    raise
        finally:
            self._status_monitoring_active = False

        return completed_timings

    def _run_with_dependencies(self) -> list[ProcessTiming]:
        """Execute processes respecting dependency order."""