Make the vision preflight detect a camera-less image

bridge/test_vision_service.py

@@ -83,7 +83,7 @@ def test_hash_pinned_yunet_model_loads_and_rejects_blank_frame(self) -> None:
             with self.assertRaises(RuntimeError):
                 verify_yunet_model(bad_model)
 
-    def test_preflight_validates_local_runtime_without_fetching_or_leaking_pairing_code(self) -> None:
+    def _run_preflight(self, debug_payload):
         with tempfile.TemporaryDirectory() as directory:
             pairing = Path(directory) / "pairing.txt"
             pairing.write_text("123456\n", encoding="ascii")
@@ -95,15 +95,52 @@ def test_preflight_validates_local_runtime_without_fetching_or_leaking_pairing_c
                 str(pairing),
                 "--preflight",
             ]
-            with patch.object(sys, "argv", argv), patch("builtins.print") as emit:
+            with patch.object(sys, "argv", argv), patch("builtins.print") as emit, patch.object(
+                CameraVisionService, "_get", **debug_payload
+            ):
                 result = main()
+        return result, json.loads(emit.call_args.args[0])
+
+    def test_preflight_reports_ready_without_leaking_pairing_code(self) -> None:
+        result, payload = self._run_preflight(
+            {"return_value": b'{"compiled_enable_camera":1,"compiled_enable_camera_host_vision":1}'}
+        )
 
-        payload = json.loads(emit.call_args.args[0])
         self.assertEqual(0, result)
         self.assertTrue(payload["ready"])
         self.assertFalse(payload["raw_frame_persistence"])
         self.assertNotIn("123456", json.dumps(payload))
 
+    def test_preflight_refuses_a_camera_less_image(self) -> None:
+        # F3's signature failure: the worker runs happily against firmware with the
+        # camera compiled out and returns zero detections forever, which is
+        # indistinguishable from a detector that simply sees nobody.
+        result, payload = self._run_preflight(
+            {"return_value": b'{"compiled_enable_camera":0,"compiled_enable_camera_host_vision":0}'}
+        )
+
+        self.assertEqual(2, result)
+        self.assertFalse(payload["ready"])
+        self.assertIn("firmware-camera-disabled", payload["reason"])
+        self.assertIn("compiled_enable_camera", payload["reason"])
+
+    def test_preflight_refuses_an_unreachable_robot(self) -> None:
+        result, payload = self._run_preflight(
+            {"side_effect": urllib.error.URLError("offline")}
+        )
+
+        self.assertEqual(2, result)
+        self.assertFalse(payload["ready"])
+        self.assertIn("robot-unreachable", payload["reason"])
+
+    def test_preflight_treats_a_truncated_debug_response_as_unknown_not_disabled(self) -> None:
+        # /debug truncates by omitting fields rather than zeroing them, so an
+        # absent flag must not be read as "camera disabled".
+        result, payload = self._run_preflight({"return_value": b'{"bridge_state":"ready"}'})
+
+        self.assertEqual(0, result)
+        self.assertTrue(payload["ready"])
+
     def test_camera_service_retries_one_transport_miss_and_records_recovery(self) -> None:
         class Detector:
             @staticmethod

bridge/vision_service.py

@@ -132,6 +132,16 @@ def encode_face_targets(pairing_code: str, faces: list[FaceTarget]) -> str:
     return f"/vision-target?p={pairing_code}&f={';'.join(encoded)}"
 
 
+def _truthy_flag(value: object) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes"}
+    return False
+
+
 def verify_yunet_model(path: str | Path) -> Path:
     model_path = Path(path).resolve()
     try:
@@ -215,6 +225,36 @@ def _get_with_retry(self, path: str, timeout: float) -> bytes:
                 time.sleep(0.05)
         raise RuntimeError("vision transport retry exhausted")
 
+    def camera_capability(self, timeout: float = 4.0) -> tuple[bool, str]:
+        """Report whether the installed firmware can serve host vision at all.
+
+        F3 is that vision delivers nothing, and the failure is silent: a worker
+        started against a camera-less image polls forever and returns zero
+        detections, which looks identical to a detector that cannot see anyone.
+        The compiled flags distinguish those two cases immediately, and /debug is
+        the unauthenticated surface, so this needs no pairing code.
+        """
+        try:
+            payload = self._get("/debug", timeout)
+        except (OSError, RuntimeError, urllib.error.URLError) as exc:
+            return False, f"robot-unreachable: {type(exc).__name__}"
+        try:
+            debug = json.loads(payload.decode("utf-8", "replace"))
+        except (UnicodeDecodeError, ValueError):
+            return False, "robot-debug-unparsable"
+        if not isinstance(debug, dict):
+            return False, "robot-debug-not-an-object"
+        missing = [
+            flag
+            for flag in ("compiled_enable_camera", "compiled_enable_camera_host_vision")
+            # A truncated /debug response omits fields rather than zeroing them, so
+            # an absent flag is unknown, not disabled. Only an explicit 0 disables.
+            if flag in debug and not _truthy_flag(debug[flag])
+        ]
+        if missing:
+            return False, "firmware-camera-disabled: " + ",".join(missing)
+        return True, "ok"
+
     def step(self, timeout: float = 4.0) -> list[FaceTarget]:
         try:
             started = time.perf_counter()
@@ -262,19 +302,21 @@ def main() -> int:
     )
     service = CameraVisionService(args.robot_url, pairing_code, OpenCvYuNetDetector(args.model_path))
     if args.preflight:
+        capable, reason = service.camera_capability()
         print(
             json.dumps(
                 {
                     "schema": "stackchan.local-vision-preflight.v1",
-                    "ready": True,
+                    "ready": capable,
+                    "reason": reason,
                     "robot_url": service.robot_url,
                     "model_sha256": YUNET_MODEL_SHA256,
                     "raw_frame_persistence": False,
                 },
                 separators=(",", ":"),
             )
         )
-        return 0
+        return 0 if capable else 2
     started = time.monotonic()
     exit_code = 0
     try: