Stream continuous-sensor changes to the live activity log

src/hatty/controllers/connection.py

@@ -218,9 +218,10 @@ def _handle_event_message(self, msg: dict) -> None:
         app.graph_ctl.record_state(new_state)
         app.notify_ctl.handle_state_change(entity_id, old_state, new_state)
         # While a logbook/event_stream subscription is active, it already
-        # carries this same state change (plus device events state_changed
-        # can never see) — appending here too would double the line (issue #19).
-        app.log_ctl.handle_state_change(entity_id, new_state)
+        # carries this same state change for most entities (issue #19) —
+        # except continuous sensors, which HA's stream excludes just like
+        # its logbook fetch does (issue #50); log_ctl decides who needs this.
+        app.log_ctl.handle_state_change(entity_id, new_state, old_state)
         app._clear_pending_call(entity_id)
         if app._detail_entity_id == entity_id:
             app.call_later(app.graph_ctl.refresh_detail_panel)

src/hatty/controllers/logbook.py

@@ -561,19 +561,28 @@ def handle_stream_frame(self, raw_entries: list[dict]) -> None:
         for entry in self.normalize(raw_entries):
             self._app.call_later(panel.add_log_entry, entry)
 
-    def handle_state_change(self, entity_id: str, new_state: Entity) -> None:
+    def handle_state_change(self, entity_id: str, new_state: Entity, old_state: "Entity | None" = None) -> None:
         """The state_changed fallback (issue #19) — while a logbook/
         event_stream subscription is active, it already carries this same
         state change (plus device events state_changed can never see), so
-        appending here too would double the line; only fires when no
-        subscription is live for the session that would want this entity."""
+        appending here too would double the line for most entities. Continuous
+        sensors are the exception (issue #50): HA's logbook stream excludes
+        them just like logbook/get_events does, so they need this fallback
+        even while a subscription is live. old_state (when known) drops a
+        no-op change — the stream never sends one, and continuous sensors
+        fire state_changed on attribute-only updates constantly."""
         session = self.live_session()
         if session is None or entity_id not in session.entity_ids:
             return
-        if self._app.client.logbook_subscription_id is not None:
+        attributes = new_state.get("attributes", {})
+        continuous = is_continuous_sensor(entity_id, attributes)
+        if self._app.client.logbook_subscription_id is not None and not continuous:
+            return
+        if old_state is not None and old_state.get("state") == new_state.get("state"):
             return
         panel = session.panel()
-        device_class = new_state.get("attributes", {}).get("device_class") or ""
+        device_class = attributes.get("device_class") or ""
+        unit = attributes.get("unit_of_measurement") or ""
         raw = {
             "when": datetime.now(timezone.utc).isoformat(),
             "state": new_state.get("state", ""),
@@ -584,5 +593,5 @@ def handle_state_change(self, entity_id: str, new_state: Entity) -> None:
         # empty — resolve_name short-circuits on it (issue #25's transport
         # consistency: this shares format_log_line/state_detail with the
         # fetched path instead of writing a raw, unlabeled string).
-        entry = normalize_entry(raw, {}, {}, {entity_id: device_class})
+        entry = normalize_entry(raw, {}, {}, {entity_id: device_class}, {entity_id: unit})
         self._app.call_later(panel.add_log_entry, entry)

tests/test_activity_log.py

@@ -699,6 +699,39 @@ async def test_live_logbook_stream_event_appends_to_activity_log(make_app):
         assert log_widget.line_count == count_before + 1
 
 
+async def test_live_continuous_sensor_state_change_appends_while_stream_is_active(make_app, sample_entities):
+    """Issue #50: HA's logbook stream never carries continuous sensors (same
+    exclusion issue #29 works around for the fetched log), so the
+    state_changed fallback must still append for one even while a
+    logbook/event_stream subscription is live — unlike a plain entity, which
+    the stream already covers (see the two tests above)."""
+    app = make_app(entities=sample_entities, config_data=NO_LIST_CONFIG)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        await pilot.press("a")
+        await pilot.pause()
+        assert app.client.logbook_subscription_id is not None
+        log_widget = app.query_one("#activity_log_panel", ActivityLogPanel).query_one("#log_widget", Log)
+        count_before = log_widget.line_count
+
+        app.client.inject_state_change(
+            {
+                "entity_id": "sensor.temperature",
+                "state": "22.0",
+                "attributes": {"friendly_name": "Temperature Sensor", "unit_of_measurement": "°C"},
+                "last_changed": "2024-01-15T10:31:00.000000+00:00",
+            },
+            old_state={
+                "entity_id": "sensor.temperature",
+                "state": "21.5",
+                "attributes": {"friendly_name": "Temperature Sensor", "unit_of_measurement": "°C"},
+            },
+        )
+        await pilot.pause()
+        assert log_widget.line_count == count_before + 1
+        assert any("Temperature Sensor → 22.0 °C" in line for line in log_widget.lines)
+
+
 async def test_logbook_stream_dedupes_a_reinjected_entry(make_app):
     """Guards the fetch/stream boundary overlap (issue #19): the same entry
     arriving twice (once via load_history, once via the live stream) renders

tests/unit/test_connection_controller.py

@@ -60,8 +60,8 @@ def __init__(self):
     def resubscribe_after_reconnect(self):
         self.reconnect_resubscribes += 1
 
-    def handle_state_change(self, entity_id, new_state):
-        self.state_changes.append((entity_id, new_state))
+    def handle_state_change(self, entity_id, new_state, old_state=None):
+        self.state_changes.append((entity_id, new_state, old_state))
 
     def handle_stream_frame(self, raw_entries):
         self.stream_frames.append(raw_entries)
@@ -300,7 +300,7 @@ def test_event_upserts_entity_and_clears_pending():
     assert app.cleared_pending == ["switch.fan"]
     assert app.graph_ctl.recorded == [new_state]
     assert app.refreshed_tree_entities == ["switch.fan"]
-    assert app.log_ctl.state_changes == [("switch.fan", new_state)]
+    assert app.log_ctl.state_changes == [("switch.fan", new_state, None)]
 
 
 def test_logbook_stream_event_routes_to_log_ctl():

tests/unit/test_logbook_controller.py

@@ -791,3 +791,59 @@ async def test_handle_state_change_appends_only_when_in_scope_and_unsubscribed()
     # Out-of-scope entity: still filtered even with the stream down.
     ctl.handle_state_change("light.b", {"state": "on", "attributes": {}})
     assert len(host.panel_widget.entries_added) == 1
+
+
+async def test_handle_state_change_appends_continuous_sensor_even_while_subscribed():
+    """Issue #50: HA's logbook stream excludes continuous sensors, so the
+    state_changed fallback must still fire for them even while a stream
+    subscription is live — unlike a plain light, which the stream covers."""
+    ctl, app = _controller()
+    host = _StubHost()
+    options = [ctl.base_option("a", "a", ["sensor.temp", "light.a"], with_devices=False)]
+    ctl.open(host, options=options, option_id="a", hint="")
+    await app.run_spawned()
+    assert app.client.logbook_subscription_id is not None
+
+    ctl.handle_state_change(
+        "sensor.temp", {"state": "21.5", "attributes": {"unit_of_measurement": "°C"}}
+    )
+    assert len(host.panel_widget.entries_added) == 1
+    assert host.panel_widget.entries_added[0]["detail"] == "21.5 °C"
+
+    # A plain light in the same session is still suppressed while subscribed.
+    ctl.handle_state_change("light.a", {"state": "on", "attributes": {}})
+    assert len(host.panel_widget.entries_added) == 1
+
+
+async def test_handle_state_change_suppresses_unit_less_sensor_while_subscribed():
+    """A sensor with no unit/state_class isn't "continuous" (issue #29's
+    predicate), so it stays covered by the stream like any other domain."""
+    ctl, app = _controller()
+    host = _StubHost()
+    options = [ctl.base_option("a", "a", ["sensor.plain"], with_devices=False)]
+    ctl.open(host, options=options, option_id="a", hint="")
+    await app.run_spawned()
+    assert app.client.logbook_subscription_id is not None
+
+    ctl.handle_state_change("sensor.plain", {"state": "ready", "attributes": {}})
+    assert host.panel_widget.entries_added == []
+
+
+async def test_handle_state_change_drops_no_op_change_when_old_state_known():
+    """Continuous sensors fire state_changed on attribute-only updates too;
+    when the caller supplies old_state, an unchanged state must not append,
+    but a genuine change still does."""
+    ctl, app = _controller()
+    host = _StubHost()
+    options = [ctl.base_option("a", "a", ["sensor.temp"], with_devices=False)]
+    ctl.open(host, options=options, option_id="a", hint="")
+    await app.run_spawned()
+
+    old = {"state": "21.5", "attributes": {"unit_of_measurement": "°C"}}
+    same = {"state": "21.5", "attributes": {"unit_of_measurement": "°C", "extra": True}}
+    ctl.handle_state_change("sensor.temp", same, old)
+    assert host.panel_widget.entries_added == []
+
+    changed = {"state": "21.6", "attributes": {"unit_of_measurement": "°C"}}
+    ctl.handle_state_change("sensor.temp", changed, old)
+    assert len(host.panel_widget.entries_added) == 1