qemu: restart the guest if its kernel panics during boot

score/itf/core/process/console.py

@@ -24,14 +24,16 @@
 
 
 class Console:
-    def __init__(self, name, reader, writer, print_logger=True, logfile=None):
+    def __init__(self, name, reader, writer, print_logger=True, logfile=None, expr_cbks=None):
         """Initializes the Console instance.
 
         :param str name: Name of the console.
         :param callable reader: Function to read lines from the console.
         :param callable writer: Function to write commands to the console.
         :param bool print_logger: Flag to enable or disable logging to the console. Defaults to True.
         :param str logfile: Path to the log file for logging output. Defaults to None.
+        :param expr_cbks: Optional list of ``(expr, callback)`` pairs registered before the reader
+            thread starts, so that no output line can be missed.
         """
         self.name = name
         self.writer = writer
@@ -41,6 +43,8 @@ def __init__(self, name, reader, writer, print_logger=True, logfile=None):
             print_logger=print_logger,
             logfile=logfile,
         )
+        for expr, cbk in expr_cbks or []:
+            self.line_reader.add_expr_cbk(expr, cbk)
         self.line_reader.start()
 
     @property
@@ -148,6 +152,7 @@ def __init__(
         linefeed: str = "\n",
         logfile: Optional[str] = None,
         print_logger: bool = True,
+        expr_cbks=None,
     ) -> None:
         """Initializes the PipeConsole instance.
 
@@ -157,6 +162,7 @@ def __init__(
         :param str linefeed: Linefeed character(s) to use when sending commands. Defaults to '\n'.
         :param Optional[str] logfile: Path to the log file for logging output. Defaults to None.
         :param bool print_logger: Flag to enable or disable logging to the console. Defaults to True.
+        :param expr_cbks: Optional list of ``(expr, callback)`` pairs, see :class:`Console`.
         """
         self._timeout = timeout
         self._linefeed = linefeed
@@ -195,7 +201,7 @@ def writer(command: str) -> None:
                 self._process.stdin.write(try_to_encode(command + "\n"))
                 self._process.stdin.flush()
 
-        super().__init__(name, reader, writer)
+        super().__init__(name, reader, writer, expr_cbks=expr_cbks)
 
 
 class LineReader(threading.Thread):

score/itf/plugins/qemu/BUILD

@@ -26,6 +26,7 @@ py_library(
     name = "qemu",
     srcs = [
         "__init__.py",
+        "boot.py",
         "checks.py",
         "image_format.py",
         "qemu.py",

score/itf/plugins/qemu/__init__.py

@@ -18,6 +18,7 @@
 import pytest
 
 from score.itf.plugins.qemu.qemu_target import qemu_target
+from score.itf.plugins.qemu.boot import boot_until_ready
 from score.itf.plugins.qemu.checks import pre_tests_phase
 from score.itf.plugins.qemu.image_format import get_image_format
 from score.itf.core.utils import padder
@@ -55,6 +56,13 @@ def _create_overlay(base_image: str) -> str:
 
 
 def pytest_addoption(parser):
+    parser.addoption(
+        "--qemu-boot-attempts",
+        action="store",
+        type=int,
+        default=3,
+        help="Maximum number of times the guest is booted if its kernel panics during boot.",
+    )
     parser.addoption(
         "--qemu-config",
         action="store",
@@ -138,7 +146,7 @@ def target_init(config, request, dlt):
                 qemu_disk=disk_overlay_path,
             )
         ) as qemu:
-            pre_tests_phase(qemu)
+            boot_until_ready(qemu, request.config.getoption("qemu_boot_attempts"), pre_tests_phase)
             yield qemu
     finally:
         if overlay_path and os.path.exists(overlay_path):

score/itf/plugins/qemu/boot.py

@@ -0,0 +1,31 @@
+# *******************************************************************************
+# Copyright (c) 2025 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+def boot_until_ready(target, max_attempts: int, run_checks):
+    """Run the pre-test checks, rebooting the guest if its kernel panicked while booting.
+
+    Only guest kernel panics are retried. Any other failure is raised immediately.
+    """
+    for attempt in range(1, max_attempts + 1):
+        try:
+            run_checks(target)
+            return
+        except Exception:
+            if not target.guest_panicked or attempt == max_attempts:
+                raise
+            logger.warning(f"Guest kernel panicked during boot (attempt {attempt}/{max_attempts}), restarting QEMU")
+            target.restart()

score/itf/plugins/qemu/qemu_process.py

@@ -12,12 +12,17 @@
 # *******************************************************************************
 import logging
 import subprocess
+import threading
 
 from score.itf.core.process.console import PipeConsole
 from score.itf.plugins.qemu.qemu import Qemu
 
 logger = logging.getLogger(__name__)
 
+# Printed by the QNX kernel on a fatal kernel error, e.g.
+# "Shutdown[1] S/C/F=5/4/3 ...". The guest does not recover afterwards.
+GUEST_PANIC_MARKER = "Shutdown["
+
 
 class QemuProcess:
     def __init__(
@@ -53,6 +58,7 @@ def __init__(
             disk=self._disk,
         )
         self._console = None
+        self._guest_panicked = threading.Event()
 
     def __enter__(self):
         return self.start()
@@ -76,10 +82,28 @@ def start(self):
             "stderr": subprocess.STDOUT,
         }
         # pylint: disable=too-many-function-args
+        self._guest_panicked.clear()
         qemu_subprocess = self._qemu.start(subprocess_params)
-        self._console = PipeConsole("QEMU", qemu_subprocess)
+        self._console = PipeConsole(
+            "QEMU",
+            qemu_subprocess,
+            expr_cbks=[(GUEST_PANIC_MARKER, lambda: self._on_guest_panic(qemu_subprocess))],
+        )
         return self
 
+    def _on_guest_panic(self, qemu_subprocess):
+        logger.error("Guest kernel panic detected on the serial console")
+        self._guest_panicked.set()
+        # The halted guest never recovers. Terminating QEMU makes any pending SSH
+        # connection fail immediately instead of waiting for a banner that never arrives.
+        if qemu_subprocess.poll() is None:
+            qemu_subprocess.terminate()
+
+    @property
+    def guest_panicked(self) -> bool:
+        """True if the guest kernel reported a panic since the last (re)start."""
+        return self._guest_panicked.is_set()
+
     def stop(self):
         logger.info("Stopping Qemu...")
         self._qemu.stop()

score/itf/plugins/qemu/qemu_target.py

@@ -126,6 +126,11 @@ def restart_process(self):
     def restart(self) -> None:
         self.restart_process()
 
+    @property
+    def guest_panicked(self) -> bool:
+        """True if the guest kernel panicked since the last (re)start of QEMU."""
+        return bool(self._process and self._process.guest_panicked)
+
     def execute(self, command: str):
         timeout = 30
         max_exec_time = 180

test/unit/BUILD

@@ -46,12 +46,20 @@ py_itf_unittest(
     deps = ["//score/itf/plugins/qemu"],
 )
 
+py_itf_unittest(
+    name = "test_qemu_boot_retry",
+    srcs = ["test_qemu_boot_retry.py"],
+    target_compatible_with = ["@platforms//os:linux"],
+    deps = ["//score/itf/plugins/qemu"],
+)
+
 test_suite(
     name = "unit",
     tests = [
         ":test_ping",
         ":test_qemu",
         ":test_qemu_auth",
+        ":test_qemu_boot_retry",
         ":test_qemu_config_schema",
         ":test_qemu_image_format",
     ],

test/unit/test_qemu_boot_retry.py

@@ -0,0 +1,97 @@
+# *******************************************************************************
+# Copyright (c) 2026 Contributors to the Eclipse Foundation
+#
+# See the NOTICE file(s) distributed with this work for additional
+# information regarding copyright ownership.
+#
+# This program and the accompanying materials are made available under the
+# terms of the Apache License Version 2.0 which is available at
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# SPDX-License-Identifier: Apache-2.0
+# *******************************************************************************
+
+import subprocess
+import sys
+import time
+
+import pytest
+
+from score.itf.plugins.qemu.boot import boot_until_ready
+from score.itf.plugins.qemu.qemu_process import QemuProcess
+
+
+class FakeTarget:
+    def __init__(self, panics_on_attempts):
+        self._panics_on_attempts = panics_on_attempts
+        self.boots = 1
+        self.guest_panicked = False
+        self.restarts = 0
+
+    def run_checks(self):
+        self.guest_panicked = self.boots in self._panics_on_attempts
+        if self.guest_panicked:
+            raise AssertionError("ssh failed")
+
+    def restart(self):
+        self.restarts += 1
+        self.boots += 1
+        self.guest_panicked = False
+
+
+@pytest.fixture
+def target():
+    def make(panics_on_attempts):
+        return FakeTarget(panics_on_attempts)
+
+    return make
+
+
+def test_no_restart_when_boot_succeeds(target):
+    fake = target(panics_on_attempts=[])
+    boot_until_ready(fake, max_attempts=3, run_checks=lambda t: t.run_checks())
+    assert fake.restarts == 0
+
+
+def test_restarts_after_guest_panic(target):
+    fake = target(panics_on_attempts=[1, 2])
+    boot_until_ready(fake, max_attempts=3, run_checks=lambda t: t.run_checks())
+    assert fake.restarts == 2
+
+
+def test_raises_when_every_attempt_panics(target):
+    fake = target(panics_on_attempts=[1, 2, 3])
+    with pytest.raises(AssertionError):
+        boot_until_ready(fake, max_attempts=3, run_checks=lambda t: t.run_checks())
+    assert fake.restarts == 2
+
+
+def test_failure_without_panic_is_not_retried():
+    fake = FakeTarget(panics_on_attempts=[])
+
+    def failing_checks(_target):
+        raise AssertionError("not a panic")
+
+    with pytest.raises(AssertionError):
+        boot_until_ready(fake, max_attempts=3, run_checks=failing_checks)
+    assert fake.restarts == 0
+
+
+def test_console_panic_marker_flags_process_and_terminates_qemu(monkeypatch):
+    monkeypatch.setattr("score.itf.plugins.qemu.qemu.Qemu._Qemu__check_qemu_is_installed", lambda self: None)
+    script = "import time; print('Shutdown[1] S/C/F=5/4/3 muon', flush=True); time.sleep(60)"
+    monkeypatch.setattr(
+        "score.itf.plugins.qemu.qemu.Qemu._Qemu__build_qemu_command",
+        lambda self: [sys.executable, "-c", script],
+    )
+    process = QemuProcess(None, "1G", 1, [], [], "pc-x86_64", None, None, None)
+    process.start()
+    try:
+        deadline = time.time() + 10
+        while not process.guest_panicked and time.time() < deadline:
+            time.sleep(0.05)
+        assert process.guest_panicked
+        assert process._qemu._subprocess.wait(timeout=10) is not None
+    finally:
+        if process._qemu._subprocess.poll() is None:
+            process._qemu._subprocess.kill()