device: do not treat a target idle at its prompt as a busy loop

CHANGES.md

@@ -3,8 +3,13 @@
 ## Unreleased
 
 - Add Renode device support.
+- Fail the build when a Kconfig override is not applied.
+- Report a slow command as TIMEOUT instead of a flood.
+- Fix slow match_all lookups on large console output.
+- Add exclude_test to session.json to skip tests per board.
 - Refuse commands longer than the target CONFIG_LINE_MAX.
 - Reopen serial ports that disappear on target reset (USB re-enumeration).
+- Do not treat a target idle at its prompt as a busy loop.
 
 ## 0.0.3
 

Documentation/usage.rst

@@ -216,6 +216,11 @@ consecutive failures the device is marked ``BUSY_LOOP``.
        interval: 60    # seconds between probes, minimum 30
        threshold: 3    # consecutive failures to declare busyloop
 
+Busy-loop detection also fires when the device prints nothing for 180
+seconds while a command is pending. A device that already printed its
+prompt is waiting for input, so silence after the prompt only makes the
+command time out and does not trigger a reboot.
+
 Device Recovery
 ===============
 

src/ntfc/device/common.py

@@ -102,6 +102,15 @@ def _output_tail_append(self, data: bytes) -> None:
         if len(self._output_tail_buf) > self._OUTPUT_TAIL_MAX:
             del self._output_tail_buf[: -self._OUTPUT_TAIL_MAX]
 
+    def _idle_at_prompt(self) -> bool:
+        """Return True when the last output ended with the shell prompt.
+
+        A target that printed its prompt is waiting for input, not stuck,
+        so silence after it is a command timeout rather than a busy loop.
+        """
+        tail = _ANSI_ESCAPE_RE.sub(b"", bytes(self._output_tail_buf[-256:]))
+        return tail.rstrip().endswith(self._dev.prompt.strip())
+
     def output_tail(self) -> str:
         """Return the most recent raw device output.
 
@@ -195,10 +204,12 @@ def _read_all(self, timeout: float = 1.0) -> bytes:
                 break
 
             # check for busy loop
-            # trigger an error if there was no data to read for a long time
+            # trigger an error if there was no data to read for a long time,
+            # unless the target is simply waiting at its prompt
             if not chunk:
-                if (  # pragma: no cover
+                if (
                     not self._state_mgr.is_busy_loop()
+                    and not self._idle_at_prompt()
                     and self._state_mgr.check_busy_loop_timeout()
                 ):
                     self._log_device_event("fault detected: busyloop")

tests/device/test_common.py

@@ -165,6 +165,7 @@ def test_device_common_slow_output_is_timeout_not_flood():
 
         config = mockdevice.return_value
         config.read_poll_interval = 0.05
+        config.prompt = "nsh> "
         dev = DeviceMock(config)
         dev._dev_is_health_priv = lambda: True
 
@@ -277,6 +278,44 @@ def test_device_common_read_until_pattern_uses_configured_poll_interval():
         read_all.assert_called_once_with(0.001)
 
 
+class DeviceMockQuickBusyloop(DeviceMock):
+    """Device mock with a short busy-loop threshold and scripted reads."""
+
+    _BUSY_LOOP_TIMEOUT = 0.2
+
+    def __init__(self, config, reads):
+        self._reads = list(reads)
+        config.prompt = "nsh> "
+        DeviceMock.__init__(self, config)
+
+    def _read(self, _=0):
+        return self._reads.pop(0) if len(self._reads) > 1 else self._reads[0]
+
+
+def test_device_common_idle_at_prompt_is_not_busyloop():
+    """A target waiting at its prompt is idle, not stuck."""
+
+    with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+        dev = DeviceMockQuickBusyloop(
+            mockdevice.return_value, [b"nsh> cmd\r\nout\r\nnsh> ", b""]
+        )
+        ret = dev.read_until_pattern(b"PASS", 1)
+        assert ret.status == CmdStatus.TIMEOUT
+        assert dev.busyloop is False
+
+
+def test_device_common_silence_without_prompt_is_busyloop():
+    """No prompt and no output past the threshold is a busy loop."""
+
+    with patch("ntfc.envconfig.EnvConfig") as mockdevice:
+        dev = DeviceMockQuickBusyloop(
+            mockdevice.return_value, [b"nsh> cmd\r\nrunning", b""]
+        )
+        ret = dev.read_until_pattern(b"PASS", 1)
+        assert ret.status == CmdStatus.TIMEOUT
+        assert dev.busyloop is True
+
+
 def test_device_common_panic_char():
 
     with patch("ntfc.device.common.get_os") as mock_get_os: