Fix buffer timespan requirement updates and overflow handling

This fixes two related issues with buffer timespan management:

1. Simplify _update_buffer_requirements to work with the updated
   get_required_timespan() contract (now returns float, not float | None).
   This prevents stale timespan requirements from persisting when all
   extractors with timespan requirements are removed.

2. Fix TemporalBuffer overflow when timespan=0.0 by properly dropping
   all existing data to make room for new values. Previously, the buffer
   would fail with "exceeds buffer capacity even after trimming" because
   trimming did nothing when timespan <= 0.

Added test_timespan_zero_trims_all_old_data_on_overflow to verify the
fix using TDD approach.

Original prompt: Consider potential bug in _update_buffer_requirements: If extractor was removed and only LatestValueExtractor remains then timespans will be empty so an old requirement for a timespan will not be cleared? Can you write a test trying to reproduce this?

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: SimonHeybrock

src/ess/livedata/dashboard/temporal_buffer_manager.py

@@ -244,11 +244,8 @@ def _update_buffer_requirements(
             List of extractors to gather requirements from.
         """
         # Compute maximum required timespan
-        timespans = [
-            ts for e in extractors if (ts := e.get_required_timespan()) is not None
-        ]
-        if timespans:
-            max_timespan = max(timespans)
+        if extractors:
+            max_timespan = max(e.get_required_timespan() for e in extractors)
             buffer.set_required_timespan(max_timespan)
             logger.debug(
                 "Set buffer required timespan to %.2f seconds (from %d extractors)",

src/ess/livedata/dashboard/temporal_buffers.py

@@ -371,7 +371,14 @@ def _initialize_buffers(self, data: sc.DataArray) -> None:
 
     def _trim_to_timespan(self, new_data: sc.DataArray) -> None:
         """Trim buffer to keep only data within required timespan."""
-        if self._required_timespan <= 0:
+        if self._required_timespan < 0:
+            return
+
+        if self._required_timespan == 0.0:
+            # Keep only the latest value - drop all existing data
+            drop_count = self._data_buffer.size
+            self._data_buffer.drop(drop_count)
+            self._time_buffer.drop(drop_count)
             return
 
         # Get latest time from new data

tests/dashboard/temporal_buffers_test.py

@@ -406,6 +406,55 @@ def test_capacity_exceeded_even_after_trimming_raises(self):
         with pytest.raises(ValueError, match="exceeds buffer capacity even after"):
             buffer.add(large_data)
 
+    def test_timespan_zero_trims_all_old_data_on_overflow(self):
+        """Test that timespan=0.0 trims all data to make room for new data."""
+        buffer = TemporalBuffer()
+        buffer.set_required_timespan(0.0)  # Keep only latest value
+        buffer.set_max_memory(100)  # Small memory limit to force overflow
+
+        # Add first data point
+        data1 = sc.DataArray(
+            sc.array(dims=['x'], values=[1.0, 2.0], unit='counts'),
+            coords={
+                'x': sc.arange('x', 2, unit='m'),
+                'time': sc.scalar(0.0, unit='s'),
+            },
+        )
+        buffer.add(data1)
+        initial_capacity = buffer._data_buffer.max_capacity
+
+        # Fill buffer to capacity
+        for t in range(1, initial_capacity):
+            data = sc.DataArray(
+                sc.array(dims=['x'], values=[float(t), float(t)], unit='counts'),
+                coords={
+                    'x': sc.arange('x', 2, unit='m'),
+                    'time': sc.scalar(float(t), unit='s'),
+                },
+            )
+            buffer.add(data)
+
+        # Buffer is now full, verify it has all data
+        result = buffer.get()
+        assert result.sizes['time'] == initial_capacity
+
+        # Add one more data point - should trigger trimming
+        # With timespan=0.0, should drop ALL old data to make room
+        data_new = sc.DataArray(
+            sc.array(dims=['x'], values=[999.0, 999.0], unit='counts'),
+            coords={
+                'x': sc.arange('x', 2, unit='m'),
+                'time': sc.scalar(999.0, unit='s'),
+            },
+        )
+        buffer.add(data_new)  # Should not raise
+
+        # Should only have the latest value
+        result = buffer.get()
+        assert result.sizes['time'] == 1
+        assert result.coords['time'].values[0] == 999.0
+        assert result['time', 0].values[0] == 999.0
+
 
 class TestVariableBuffer:
     """Tests for VariableBuffer."""