Preserve native USB failures when endpoint cleanup also fails

test/test_official_console_acceptance.py

@@ -1,4 +1,5 @@
 """Offline owned-CDP startup regressions; never open a browser or serial port."""
+from contextlib import contextmanager
 import importlib.util
 import io
 import json
@@ -260,6 +261,132 @@ class OwnedCleanupTests(unittest.TestCase):
         self.assertIs(signal.getsignal(signal.SIGTERM), previous)
 
 
+class PrimaryFailureCleanupTests(unittest.TestCase):
+    def main_with(self, execute):
+        with tempfile.TemporaryDirectory() as directory:
+            work = Path(directory)
+            work.chmod(0o700)
+            output = work / "result.json"
+            stdout = io.StringIO()
+            with patch.object(acceptance, "execute", side_effect=execute), \
+                    patch("sys.stdout", new=stdout):
+                status = acceptance.main(["--serial", "exact", "--expected-board", "board",
+                    "--expected-version", "version", "--logging", "on", "--console-source", "/unused",
+                    "--artifact-manifest", "/unused", "--output", str(output),
+                    "--work-dir", str(work), "--execute"])
+            self.assertEqual(output.stat().st_mode & 0o777, 0o600)
+            return status, json.loads(output.read_text()), stdout.getvalue()
+
+    @staticmethod
+    @contextmanager
+    def disappeared_tty(_port):
+        try:
+            yield {"vmin_during_test": 1, "vtime_during_test": 0}
+        finally:
+            raise FileNotFoundError("private missing /dev/ttyACM-secret")
+
+    def test_idle_disconnect_survives_missing_tty_cleanup_and_main_report(self):
+        original = acceptance.NativeConsoleJavaScriptError({"exception": {
+            "className": "Error", "description": "Error: console_disconnected_while_idle\nprivate stack"}})
+
+        def disconnected(_args, report):
+            report["phase"] = "uninterrupted_idle"
+            with acceptance.recorded_linux_web_serial("/dev/inert", report):
+                raise original
+
+        with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty):
+            status, report, stdout = self.main_with(disconnected)
+        self.assertEqual(status, 1)
+        self.assertFalse(report["ok"])
+        self.assertEqual(report["error_type"], "NativeConsoleJavaScriptError")
+        self.assertEqual(report["error"], "native_console_javascript_failed:console_disconnected_while_idle")
+        self.assertEqual(report["primary_failure"]["phase"], "uninterrupted_idle")
+        self.assertEqual(report["TTY_cleanup_failure"]["class"], "FileNotFoundError")
+        self.assertIn("message_sha256", report["TTY_cleanup_failure"])
+        self.assertEqual(report["TTY_cleanup_endpoint"], "missing")
+        self.assertFalse(report["original_TTY_restored"])
+        self.assertNotIn("private", json.dumps(report) + stdout)
+
+    def test_cleanup_failure_after_successful_body_cannot_pass(self):
+        def completed(_args, report):
+            report["phase"] = "complete"
+            with acceptance.recorded_linux_web_serial("/dev/inert", report):
+                report["native_streams_closed"] = True
+
+        with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty):
+            status, report, _stdout = self.main_with(completed)
+        self.assertEqual(status, 1)
+        self.assertFalse(report["ok"])
+        self.assertEqual(report["error_type"], "FileNotFoundError")
+        self.assertNotIn("primary_failure", report)
+        self.assertFalse(report["original_TTY_restored"])
+
+    def test_browser_and_tty_cleanup_failures_cannot_replace_original_transport_error(self):
+        original = TimeoutError("private original transport deadline")
+        report = {"phase": "uninterrupted_idle"}
+
+        def browser_cleanup():
+            raise OSError("private browser cleanup error")
+
+        with patch.object(acceptance, "prepared_linux_web_serial", self.disappeared_tty), \
+                self.assertRaises(TimeoutError) as caught:
+            with acceptance.recorded_linux_web_serial("/dev/inert", report):
+                with acceptance.preserving_cleanup_failure(report, "browser_cleanup_failure", browser_cleanup):
+                    raise original
+        self.assertIs(caught.exception, original)
+        self.assertEqual(report["primary_failure"]["class"], "TimeoutError")
+        self.assertEqual(report["browser_cleanup_failure"]["class"], "OSError")
+        self.assertEqual(report["TTY_cleanup_failure"]["class"], "FileNotFoundError")
+        self.assertNotIn("private", json.dumps(report))
+
+    def test_replaced_tty_is_not_modified_or_reported_as_restored(self):
+        # Real Linux pseudoterminals exercise the unchanged identity guard;
+        # no physical serial device or browser is opened.
+        import os
+        import pty
+        import termios
+        import linux_web_serial as tty_helper
+        master, slave = pty.openpty()
+        other_master, other_slave = pty.openpty()
+        try:
+            path, other_path = os.ttyname(slave), os.ttyname(other_slave)
+            attributes = termios.tcgetattr(other_slave)
+            attributes[6][termios.VTIME] = b"\x07"
+            termios.tcsetattr(other_slave, termios.TCSANOW, attributes)
+            original_other = termios.tcgetattr(other_slave)
+            original_open = tty_helper._open_tty
+            calls = []
+
+            def replaced_open(selected):
+                calls.append(selected)
+                return original_open(selected if len(calls) == 1 else other_path)
+
+            report = {"phase": "complete"}
+            with patch.object(tty_helper, "_open_tty", side_effect=replaced_open), \
+                    self.assertRaisesRegex(RuntimeError, "endpoint changed"):
+                with acceptance.recorded_linux_web_serial(path, report):
+                    pass
+            self.assertEqual(termios.tcgetattr(other_slave), original_other)
+            self.assertEqual(report["TTY_cleanup_endpoint"], "identity_changed")
+            self.assertFalse(report["original_TTY_restored"])
+            self.assertEqual(report["TTY_cleanup_failure"]["class"], "RuntimeError")
+            self.assertNotIn("primary_failure", report)
+        finally:
+            for fd in (master, slave, other_master, other_slave):
+                os.close(fd)
+
+    def test_unknown_runtime_error_text_is_hashed_in_final_report(self):
+        def failed(_args, _report):
+            raise RuntimeError("private password from an external dependency")
+
+        status, report, stdout = self.main_with(failed)
+        self.assertEqual(status, 1)
+        self.assertFalse(report["ok"])
+        self.assertEqual(report["error"], "hardware_acceptance_interrupted_or_failed")
+        self.assertIn("message_sha256", report["failure"])
+        self.assertNotIn("private", json.dumps(report) + stdout)
+
+
 class SafeJavaScriptFailureTests(unittest.TestCase):
     def evaluate_error(self, details):
         page = object.__new__(acceptance.CDP)

tools/hil/official_console_acceptance.py

@@ -180,6 +180,87 @@ class NativeConsoleTerminated(KeyboardInterrupt):
     """Unwind the owned resources and report a normal SIGTERM interruption."""
 
 
+NATIVE_ERROR_CODES = frozenset({
+    "CDP_browser_does_not_match_owned_profile", "CDP_request_failed", "CDP_request_deadline",
+    "USB_endpoint_changed_before_browser", "USB_identity_or_ownership_changed_after_cleanup",
+    "USB_vendor_product_ID_required", "complete_CRLF_reply_required",
+    "exact_USB_serial_requires_one_primary_interface", "existing_private_work_directory_required",
+    "firmware_command_rejected", "logging_mode_changed", "logging_mode_must_be_preconfigured",
+    "native_stream_cleanup_failed", "official_console_source_pin_mismatch",
+    "official_module_did_not_load", "owned_Chromium_exited_before_ready",
+    "owned_Chromium_startup_deadline", "owned_browser_requires_one_initial_blank_page",
+    "reopen_reset_or_uptime_regression", "runtime_board_mismatch",
+    "runtime_identity_changed_on_reopen", "runtime_version_mismatch",
+    "selected_serial_port_must_be_unowned", "uninterrupted_no_transmit_idle_required",
+    "uptime_delta_indicates_reset_or_stall", "usb_logging_reply_format_unrecognized",
+    "work_directory_must_be_private_and_owned",
+})
+NATIVE_ERROR_CLASSES = frozenset({
+    "RuntimeError", "OSError", "FileNotFoundError", "PermissionError", "TimeoutError",
+    "ConnectionError", "ConnectionResetError", "BrokenPipeError", "ValueError",
+    "KeyboardInterrupt", "NativeConsoleTerminated", "NativeConsoleJavaScriptError",
+})
+
+
+def safe_native_failure(error, phase):
+    """Keep controlled codes; hash arbitrary transport, TTY and exception text."""
+    name = type(error).__name__
+    safe = {"class": name if name in NATIVE_ERROR_CLASSES else "unrecognized",
+            "phase": phase, "code": "hardware_acceptance_interrupted_or_failed"}
+    if isinstance(error, NativeConsoleJavaScriptError):
+        safe["code"] = "native_console_javascript_failed:" + error.safe_details["code"]
+        safe["javascript_exception"] = error.safe_details
+    elif type(error) is RuntimeError and str(error) in NATIVE_ERROR_CODES:
+        safe["code"] = str(error)
+    else:
+        safe["message_sha256"] = hashlib.sha256(
+            str(error).encode("utf-8", errors="replace")).hexdigest()
+    return safe
+
+
+@contextmanager
+def preserving_cleanup_failure(report, label, cleanup):
+    """A cleanup failure must not replace the failed native trial's evidence."""
+    primary = None
+    try:
+        yield
+    except BaseException as error:
+        primary = error
+        report.setdefault("primary_failure", safe_native_failure(error, report.get("phase")))
+        raise
+    finally:
+        try:
+            cleanup()
+        except BaseException as error:
+            report[label] = safe_native_failure(error, report.get("phase"))
+            if primary is None:
+                raise
+
+
+@contextmanager
+def recorded_linux_web_serial(port, report):
+    manager = prepared_linux_web_serial(port)
+    tty = manager.__enter__()
+    report["original_TTY_restored"] = False
+
+    def restore():
+        try:
+            # This helper's exit only restores termios; the primary trial
+            # exception is handled by the surrounding cleanup boundary.
+            manager.__exit__(None, None, None)
+        except FileNotFoundError:
+            report["TTY_cleanup_endpoint"] = "missing"
+            raise
+        except RuntimeError as error:
+            if str(error) == "TTY endpoint changed; original attributes were not applied":
+                report["TTY_cleanup_endpoint"] = "identity_changed"
+            raise
+        report["original_TTY_restored"] = True
+
+    with preserving_cleanup_failure(report, "TTY_cleanup_failure", restore):
+        yield tty
+
+
 @contextmanager
 def cleanup_on_sigterm():
     previous = signal.getsignal(signal.SIGTERM)
@@ -419,9 +500,10 @@ def execute(args, report):
     browser = page = None
     browser_log = (work / "chromium-private.log").open("xb")
     try:
-        with prepared_linux_web_serial(device.device) as tty:
+        with recorded_linux_web_serial(device.device, report) as tty:
             report["TTY_preflight"] = tty
-            try:
+            with preserving_cleanup_failure(report, "browser_cleanup_failure",
+                    lambda: close_owned_browser(page, browser, report)):
                 # Re-enumerate the exact UID and ownership after TTY preparation.
                 require(select_device(args.serial).device == device.device, "USB_endpoint_changed_before_browser")
                 browser = subprocess.Popen([str(args.chromium), "--headless=new", "--no-sandbox",
@@ -495,9 +577,6 @@ def execute(args, report):
                               setup_writes=page.evaluate("hil.setup_writes"))
                 report["native_streams_closed"] = True
                 report["phase"] = "complete"
-            finally:
-                close_owned_browser(page, browser, report)
-        report["original_TTY_restored"] = True
         require(select_device(args.serial).device == device.device, "USB_identity_or_ownership_changed_after_cleanup")
         require(not (report.get("browser_stream_cleanup_error") or report.get("browser_socket_cleanup_error")),
                 "native_stream_cleanup_failed")
@@ -556,8 +635,10 @@ def main(argv=None):
         status = 0
     except BaseException as error:
         # Preserve only controlled errors, never arbitrary native console output.
-        report["error_type"] = type(error).__name__
-        report["error"] = str(error) if isinstance(error, RuntimeError) else "hardware_acceptance_interrupted_or_failed"
+        failure = safe_native_failure(error, report.get("phase"))
+        report["failure"] = failure
+        report["error_type"] = failure["class"]
+        report["error"] = failure["code"]
         if isinstance(error, NativeConsoleJavaScriptError):
             report["javascript_exception"] = error.safe_details
         if isinstance(error, NativeConsoleTerminated):