Improve instrumentation

ThomasFarstrike · ThomasFarstrike · commit 917f2c54b504 · 2026-05-19T09:41:39.000+02:00
diff --git a/AGENTS.md b/AGENTS.md
@@ -75,4 +75,5 @@ MPOS Controller (`scripts/mpos_controller.py`):
 - IMPORTANT: `get_widget_tree()` and `get_visible_text()` include ALL children of scrollable parents, including off-screen items. y1/y2 coordinates are in content space, not screen space. To know what's actually visible, combine a screenshot (`mpos.screenshot()`) with the ppq-vision skill.
 - `_read_remote_file` / `write_remote_file`: ProcessBackend uses base64 (works over PTY), SerialBackend uses `mpremote cp` (reliable over USB).
 - `mpos.screenshot()` captures via `capture_screenshot()` on device, then reads raw file and converts to BMP via `_build_bmp()`.
+- The notification bar (top status bar) is NOT always present. It's controlled by the `bar_open` global in `internal_filesystem/lib/mpos/ui/topmenu.py`. Check it with `mpos.eval("mpos.ui.topmenu.bar_open")` (the module is already imported by `main.py`). When open, its height is `AppearanceManager.NOTIFICATION_BAR_HEIGHT` (24px).
 - All tests pass covering exec, eval, screenshot, input simulation, screen introspection, file I/O, and physical device control.
diff --git a/scripts/mpos_controller.py b/scripts/mpos_controller.py
@@ -22,6 +22,7 @@
 import subprocess
 import sys
 import tempfile
+import termios
 import time
 import platform
 
@@ -304,18 +305,23 @@ def read_until(self, ending, timeout=30):
 
     def wait_for_boot(self, timeout=30):
         t0 = time.monotonic()
-        self._drain(0.3)
+        data = b""
+        ctrl_c_sent = False
         while time.monotonic() - t0 < timeout:
-            self.stream.write(b"\x03")
-            time.sleep(0.2)
-            if self._data_waiting(0.3):
-                data = self.stream.read(4096)
-                if b">>> " in data:
-                    self.stream.write(b"\n")
-                    time.sleep(0.3)
-                    self._drain(0.5)
-                    return
-        # Ctrl-C didn't yield a prompt; try Ctrl-B to enter friendly REPL
+            if self._data_waiting(0.1):
+                chunk = self.stream.read(4096)
+                if chunk:
+                    data += chunk
+                    if b">>> " in data:
+                        self._drain(0.5)
+                        self.stream.write(b"\n")
+                        time.sleep(0.2)
+                        self._drain(0.3)
+                        return
+            elif not ctrl_c_sent and time.monotonic() - t0 > 2:
+                self.stream.write(b"\x03")
+                time.sleep(0.3)
+                ctrl_c_sent = True
         self.stream.write(b"\x02")
         time.sleep(0.2)
         self._drain(0.3)
@@ -325,8 +331,8 @@ def wait_for_boot(self, timeout=30):
                 "aioREPL prompt not found.\n" + data.decode("utf-8", "replace")[-2000:]
             )
         self.stream.write(b"\n")
-        time.sleep(0.5)
-        self._drain(1.0)
+        time.sleep(0.3)
+        self._drain(0.5)
 
     def exec(self, code, timeout=30):
         """
@@ -337,9 +343,12 @@ def exec(self, code, timeout=30):
         to delimit output from REPL chatter.
         """
         self._drain(0.1)
+
+        # Enter paste mode and wait for it to engage
         self.stream.write(b"\x05")
-        time.sleep(0.05)
-        self._drain(0.1)
+        time.sleep(0.2)
+        self._drain(0.5)
+
         payload = "print('{}')\n".format(SENTINEL) + code.rstrip()
         self.stream.write(payload.encode("utf-8"))
         self.stream.write(b"\x04")
@@ -388,6 +397,21 @@ def start(self):
             pass
         time.sleep(0.3)
         master_fd, slave_fd = pty.openpty()
+        # Set master side to raw mode so control chars (Ctrl-C, Ctrl-D, Ctrl-E)
+        # pass through unmodified instead of being intercepted by the line
+        # discipline (which would consume Ctrl-D as EOF, break paste mode, etc.)
+        try:
+            iflag, oflag, cflag, lflag, ispeed, ospeed, cc = termios.tcgetattr(master_fd)
+            iflag &= ~(termios.BRKINT | termios.IGNBRK | termios.ICRNL | termios.INLCR | termios.IGNCR | termios.IXON | termios.IXOFF | termios.PARMRK | termios.INPCK | termios.ISTRIP)
+            oflag &= ~(termios.OPOST)
+            cflag &= ~(termios.PARENB | termios.CSIZE)
+            cflag |= termios.CS8
+            lflag &= ~(termios.ECHO | termios.ECHONL | termios.ICANON | termios.ISIG | termios.IEXTEN)
+            cc[termios.VMIN] = 1
+            cc[termios.VTIME] = 0
+            termios.tcsetattr(master_fd, termios.TCSANOW, [iflag, oflag, cflag, lflag, ispeed, ospeed, cc])
+        except Exception:
+            pass
         self.proc = subprocess.Popen(
             [
                 self.binary,
@@ -419,10 +443,12 @@ def stop(self):
         if self.proc:
             try:
                 os.killpg(os.getpgid(self.proc.pid), signal.SIGTERM)
-                self.proc.wait(timeout=5)
+                self.proc.wait(timeout=3)
             except Exception:
+                pass
+            if self.proc and self.proc.poll() is None:
+                self.proc.kill()
                 try:
-                    self.proc.kill()
                     self.proc.wait(timeout=2)
                 except Exception:
                     pass
@@ -434,6 +460,12 @@ def stop(self):
                 pass
             self.master_fd = None
 
+    def restart(self):
+        """Stop and restart the backend."""
+        self.stop()
+        time.sleep(0.3)
+        self.start()
+
     def __del__(self):
         self.stop()
 
