fix(client): expose safe T-103 failure stages

client/tests/pdd/test_sku_selection.py

@@ -15,7 +15,7 @@ from PIL import Image
 
 import cmbuyer_client.pdd as pdd
 import cmbuyer_client.pdd.sku_selection_runner as runner_module
-from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
+from cmbuyer_client.device.adb import AdbDevice, DeviceConnectionError, DeviceInspection
 from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
 from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
 from cmbuyer_client.pdd.sku_selection_runner import (
@@ -23,6 +23,7 @@ from cmbuyer_client.pdd.sku_selection_runner import (
     SkuSelectionRunError,
     SkuSelectionScreenshotError,
     UiautomatorSkuPanelAdapter,
+    safe_failure_stage,
 )
 
 
@@ -383,6 +384,38 @@ class SkuSelectionRunnerTests(unittest.TestCase):
             self.assertFalse((target / "hierarchy.xml").exists())
             self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
 
+    def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None:
+        class NoPanelAfterEntry(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "click":
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    return ""
+                return super().jsonrpc_call(method, params, timeout)
+
+        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError) as raised:
+            self._runner(_FakeAdb(), NoPanelAfterEntry()).run(
+                "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
+            )
+        self.assertEqual(safe_failure_stage(raised.exception), "sku_entry")
+
+        forged = SkuSelectionError("<hierarchy>private</hierarchy>")
+        setattr(forged, "_cmbuyer_failure_stage", "<private-stage>")
+        self.assertEqual(safe_failure_stage(forged), "unknown")
+
+        class HostileSetterError(DeviceConnectionError):
+            def __setattr__(self, name: str, value: object) -> None:
+                raise KeyboardInterrupt("SERIAL=private <hierarchy>secret</hierarchy>")
+
+        class FailingAdb(_FakeAdb):
+            def inspect(self, serial: str) -> DeviceInspection:
+                raise HostileSetterError("private")
+
+        with TemporaryDirectory() as temporary, self.assertRaises(HostileSetterError) as hostile:
+            SkuSelectionRunner(FailingAdb(), lambda serial: _RawDevice(), 10).run(
+                "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
+            )
+        self.assertEqual(safe_failure_stage(hostile.exception), "unknown")
+
     def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
         with TemporaryDirectory() as temporary:
             target = Path(temporary) / "result"
@@ -401,8 +434,9 @@ class SkuSelectionRunnerTests(unittest.TestCase):
     def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
         with TemporaryDirectory() as temporary:
             target = Path(temporary) / "result"
-            with self.assertRaises(SkuSelectionScreenshotError):
+            with self.assertRaises(SkuSelectionScreenshotError) as screenshot_failure:
                 self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertEqual(safe_failure_stage(screenshot_failure.exception), "screenshot_capture")
             self.assertFalse(target.exists())
             self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
 
@@ -418,8 +452,9 @@ class SkuSelectionRunnerTests(unittest.TestCase):
             target.mkdir()
             sentinel = target / "keep"
             sentinel.write_text("keep", encoding="utf-8")
-            with self.assertRaises(SkuSelectionRunError):
+            with self.assertRaises(SkuSelectionRunError) as existing_target_failure:
                 self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertEqual(safe_failure_stage(existing_target_failure.exception), "precheck")
             self.assertEqual(adb.calls, [])
             self.assertEqual(device.calls, [])
             self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
@@ -654,25 +689,43 @@ class SkuSelectionCliTests(unittest.TestCase):
     def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
         script = _load_runner_script()
 
-        class FlowFailingRunner:
-            def __init__(self, *args: object, **kwargs: object) -> None: pass
-            def run(self, *args: object, **kwargs: object) -> object:
-                raise SkuSelectionError("<hierarchy>page-body</hierarchy>")
+        secret = "SERIAL=192.168.0.173:5555 PATH=C:/private <hierarchy>page-body</hierarchy>"
 
-        stderr = BytesIO()
-        # TextIOWrapper keeps the assertion independent from host console encoding.
-        import io
-        text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
-        with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
-            status = script.main([
-                "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
-                "--size", _TASK_SIZE, "--output-dir", "evidence",
-            ])
-        text_stderr.flush()
-        output = stderr.getvalue().decode("utf-8")
-        self.assertEqual(status, 1)
-        self.assertNotIn("Traceback", output)
-        self.assertNotIn("page-body", output)
+        class HostileGetterError(SkuSelectionError):
+            def __getattribute__(self, name: str) -> object:
+                if name == "_cmbuyer_failure_stage":
+                    raise RuntimeError(secret)
+                return super().__getattribute__(name)
+
+        class HostileStage(str):
+            def __hash__(self) -> int:
+                raise RuntimeError(secret)
+
+        hostile_string = SkuSelectionError(secret)
+        setattr(hostile_string, "_cmbuyer_failure_stage", HostileStage("sku_entry"))
+
+        for failure in (SkuSelectionError(secret), HostileGetterError(secret), hostile_string):
+            class FlowFailingRunner:
+                def __init__(self, *args: object, **kwargs: object) -> None: pass
+                def run(self, *args: object, **kwargs: object) -> object:
+                    raise failure
+
+            stderr = BytesIO()
+            # TextIOWrapper keeps the assertion independent from host console encoding.
+            import io
+            text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
+            with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
+                status = script.main([
+                    "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
+                    "--size", _TASK_SIZE, "--output-dir", "evidence",
+                ])
+            text_stderr.flush()
+            output = stderr.getvalue().decode("utf-8")
+            self.assertEqual(status, 1)
+            self.assertIn("stage=unknown", output)
+            self.assertNotIn("Traceback", output)
+            self.assertNotIn(secret, output)
+            self.assertNotIn("page-body", output)
 
 
 def _load_runner_script() -> object: