Report a silent robot as gone instead of latching its last state

bridge/dashboard_service.py

@@ -90,6 +90,19 @@
     8: "Error",
 }
 
+# A robot that has stopped sending heartbeats is gone, whatever the last socket
+# event or retained /debug snapshot claimed. Both of those latch:
+# note_client_disconnected() only clears the flag when the peer host matches,
+# refresh_robot() only clears it while the bridge listener is down, and status()
+# re-asserts connectivity from a _debug snapshot that is never invalidated. A
+# robot absent for days therefore kept reporting as connected, in whatever mode
+# its final heartbeat carried.
+#
+# The window is deliberately far longer than the heartbeat cadence. One missed,
+# late, or unreadable sample must never be classified as a robot failure; only a
+# sustained absence counts.
+ROBOT_HEARTBEAT_PRESENCE_TIMEOUT_SECONDS = 30.0
+
 
 def _utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
@@ -492,6 +505,16 @@ def status(self) -> dict[str, object]:
             robot_connected = self._robot_connected or (
                 debug.get("network_state") == "connected" and debug.get("bridge_state") == "ready"
             )
+            # Heartbeat silence outranks both latched sources above. Absent any
+            # heartbeat at all (age is None) nothing is asserted either way, so a
+            # freshly accepted socket still reads as connected until it has had a
+            # chance to report.
+            if (
+                heartbeat_age is not None
+                and heartbeat_age > ROBOT_HEARTBEAT_PRESENCE_TIMEOUT_SECONDS
+            ):
+                robot_connected = False
+                mode_name = "Unknown"
             initiative = (
                 {**self.initiative_policy.status(), "available": True}
                 if self.initiative_policy is not None

bridge/test_dashboard_service.py

@@ -109,6 +109,46 @@ def test_heartbeat_status_is_allowlisted(self) -> None:
         self.assertEqual(82, status["robot"]["batteryPercent"])
         self.assertNotIn("private_text", json.dumps(status))
 
+    def test_sustained_heartbeat_silence_reports_the_robot_as_gone(self) -> None:
+        # Observed on the reference robot: the console reported connected and
+        # "Thinking" on a heartbeat 4.3 days old, with no socket at the OS level.
+        self.runtime.note_client_connected("192.168.1.238", 50123)
+        self.runtime.note_heartbeat({"type": "heartbeat", "robot_mode": 4})
+        self.runtime._record_debug({"network_state": "connected", "bridge_state": "ready"})
+
+        fresh = self.runtime.status()
+        self.assertTrue(fresh["robot"]["connected"])
+        self.assertEqual("Thinking", fresh["robot"]["mode"])
+
+        # Neither latched source is cleared here: the socket never reported a
+        # disconnect and the retained /debug snapshot still says ready. Only the
+        # heartbeat gap should decide it.
+        self.runtime._last_heartbeat_at -= 10_000.0
+
+        stale = self.runtime.status()
+        self.assertFalse(stale["robot"]["connected"])
+        self.assertEqual("Unknown", stale["robot"]["mode"])
+        self.assertFalse(stale["bridge"]["connected"])
+
+    def test_brief_heartbeat_gap_is_not_treated_as_a_robot_failure(self) -> None:
+        self.runtime.note_client_connected("192.168.1.238", 50123)
+        self.runtime.note_heartbeat({"type": "heartbeat", "robot_mode": 1})
+
+        # A few missed samples must not flip the console to disconnected.
+        self.runtime._last_heartbeat_at -= 10.0
+
+        status = self.runtime.status()
+        self.assertTrue(status["robot"]["connected"])
+        self.assertEqual("Idle", status["robot"]["mode"])
+
+    def test_connected_robot_reads_connected_before_its_first_heartbeat(self) -> None:
+        self.runtime.note_client_connected("192.168.1.238", 50123)
+
+        status = self.runtime.status()
+
+        self.assertTrue(status["robot"]["connected"])
+        self.assertIsNone(status["robot"]["heartbeatAgeSeconds"])
+
     def test_pipeline_health_attributes_failures_without_turn_content(self) -> None:
         self.runtime.note_pipeline_stage(
             "researching",