Multi session support

src/ess/livedata/dashboard/correlation_plotter.py

@@ -28,7 +28,7 @@
     PlotDisplayParams1d,
     PlotDisplayParams2d,
 )
-from .plots import ImagePlotter, LinePlotter
+from .plots import ImagePlotter, LinePlotter, PresenterBase
 
 
 class NormalizationParams(pydantic.BaseModel):
@@ -148,9 +148,10 @@ def _make_lookup(axis_data: sc.DataArray, data_max_time: sc.Variable) -> sc.bins
     mode = 'nearest' if data_max_time < axis_min_time else 'previous'
     if mode == 'nearest':  # scipp>=25.11.0 rejects int coords in this mode
         dim = axis_values.dim
-        axis_values = axis_values.assign_coords(
-            {dim: axis_values.coords[dim].astype('float64')}
-        )
+        coord = axis_values.coords[dim]
+        # Only convert to float64 if needed; datetime64 coords are accepted as-is
+        if coord.dtype != sc.DType.datetime64:
+            axis_values = axis_values.assign_coords({dim: coord.astype('float64')})
     return sc.lookup(axis_values, mode=mode)
 
 
@@ -189,7 +190,7 @@ def __init__(
     def initialize_from_data(self, data: dict[str, Any]) -> None:
         """No-op: histogram edges are computed dynamically on each call."""
 
-    def __call__(self, data: dict[str, Any]) -> Any:
+    def compute(self, data: dict[str, Any]) -> None:
         """Compute histograms for all data sources and render.
 
         Parameters
@@ -238,7 +239,26 @@ def __call__(self, data: dict[str, Any]) -> Any:
             else:
                 histograms[key] = dependent.hist(bin_spec)
 
-        return self._renderer(histograms)
+        self._renderer.compute(histograms)
+
+    def get_cached_state(self) -> Any | None:
+        """Get the last computed state from the renderer."""
+        return self._renderer.get_cached_state()
+
+    def has_cached_state(self) -> bool:
+        """Check if the renderer has computed state."""
+        return self._renderer.has_cached_state()
+
+    def create_presenter(self) -> PresenterBase:
+        """Create a presenter owned by this plotter.
+
+        Uses the renderer for presentation.
+        """
+        return self._renderer.create_presenter(owner=self)
+
+    def mark_presenters_dirty(self) -> None:
+        """Mark all presenters as dirty by delegating to the renderer."""
+        self._renderer.mark_presenters_dirty()
 
 
 class CorrelationHistogram1dPlotter(CorrelationHistogramPlotter):

src/ess/livedata/dashboard/dashboard.py

@@ -7,13 +7,15 @@
 from contextlib import ExitStack
 
 import panel as pn
-from holoviews import Dimension, streams
+from holoviews import Dimension
 
 from ess.livedata import ServiceBase
 
 from .config_store import ConfigStoreManager
 from .dashboard_services import DashboardServices
 from .kafka_transport import DashboardKafkaTransport
+from .session_registry import SessionId
+from .session_updater import SessionUpdater
 from .transport import NullTransport, Transport
 
 # Global throttling for sliders, etc.
@@ -42,17 +44,14 @@ def __init__(
         self._exit_stack = ExitStack()
         self._exit_stack.__enter__()
 
-        self._callback = None
-
         # Config store manager for file-backed persistent UI state (GUI dashboards)
         config_manager = ConfigStoreManager(instrument=instrument, store_type='file')
 
-        # Setup all dashboard services
+        # Setup all dashboard services (includes session registry)
         self._services = DashboardServices(
             instrument=instrument,
             dev=dev,
             exit_stack=self._exit_stack,
-            pipe_factory=streams.Pipe,
             transport=self._create_transport(transport),
             config_manager=config_manager,
         )
@@ -89,18 +88,19 @@ def create_sidebar_content(self) -> pn.viewable.Viewable:
         pass
 
     @abstractmethod
-    def create_main_content(self) -> pn.viewable.Viewable:
-        """Override this method to create the main dashboard content."""
-
-    def _step(self):
-        """Step function for periodic updates."""
-        # We use hold() to ensure that the UI does not update repeatedly when multiple
-        # messages are processed in a single step. This is important to avoid, e.g.,
-        # multiple lines in the same plot, or different plots updating in short
-        # succession, which is visually distracting.
-        # Furthermore, this improves performance by reducing the number of re-renders.
-        with pn.io.hold():
-            self._services.orchestrator.update()
+    def create_main_content(
+        self, session_updater: SessionUpdater
+    ) -> pn.viewable.Viewable:
+        """
+        Override this method to create the main dashboard content.
+
+        Parameters
+        ----------
+        session_updater:
+            The session updater for this browser session. Widgets that need
+            to register handlers for periodic updates should receive this
+            in their constructor.
+        """
 
     def get_dashboard_title(self) -> str:
         """Get the dashboard title. Override for custom titles."""
@@ -110,37 +110,65 @@ def get_header_background(self) -> str:
         """Get the header background color. Override for custom colors."""
         return '#2596be'
 
-    def start_periodic_updates(self, period: int = 500) -> None:
+    def _get_session_id(self) -> SessionId:
+        """Get the current session ID from Panel state."""
+        session_context = pn.state.curdoc.session_context
+        if session_context is not None:
+            return SessionId(session_context.id)
+        # Fallback for non-session contexts (e.g., testing)
+        return SessionId('unknown')
+
+    def _create_session_updater(self) -> SessionUpdater:
         """
-        Start periodic updates for the dashboard.
+        Create a SessionUpdater for the current browser session.
+
+        The updater auto-registers with the session registry.
+        """
+        return SessionUpdater(
+            session_id=self._get_session_id(),
+            session_registry=self._services.session_registry,
+            notification_queue=self._services.notification_queue,
+        )
+
+    def _start_periodic_callback(
+        self, session_updater: SessionUpdater, period: int = 500
+    ) -> None:
+        """
+        Start the periodic callback for a session.
 
         Parameters
         ----------
+        session_updater:
+            The session updater to drive with the periodic callback.
         period:
             The period in milliseconds for the periodic update step.
-            Default is 500 ms. Even if the backend produces updates, e.g., once per
-            second, this default should reduce UI lag somewhat. If there are no new
-            messages, the step function should not do anything.
         """
-        if self._callback is not None:
-            # Callback from previous session, e.g., before reloading the page. As far as
-            # I can tell the garbage collector does clean this up eventually, but
-            # let's be explicit.
-            self._callback.stop()
+        session_id = session_updater.session_id
+        session_registry = self._services.session_registry
 
         def _safe_step():
             try:
-                self._step()
+                session_updater.periodic_update()
             except Exception:
                 self._logger.exception("Error in periodic update step.")
 
-        self._callback = pn.state.add_periodic_callback(_safe_step, period=period)
-        self._logger.info("Periodic updates started")
+        callback = pn.state.add_periodic_callback(_safe_step, period=period)
+
+        def _cleanup_session(session_context):
+            self._logger.info("Session destroyed: %s", session_id)
+            session_registry.unregister(session_id)
+            callback.stop()
+
+        pn.state.on_session_destroyed(_cleanup_session)
+        self._logger.info("Periodic updates started for session %s", session_id)
 
     def create_layout(self) -> pn.template.MaterialTemplate:
         """Create the basic dashboard layout."""
+        # Create session updater first so widgets can register handlers
+        session_updater = self._create_session_updater()
+
         sidebar_content = self.create_sidebar_content()
-        main_content = self.create_main_content()
+        main_content = self.create_main_content(session_updater)
 
         template = pn.template.MaterialTemplate(
             title=self.get_dashboard_title(),
@@ -150,7 +178,7 @@ def create_layout(self) -> pn.template.MaterialTemplate:
         )
         # Inject CSS for offline mode (replaces Material Icons font with Unicode)
         template.config.raw_css.extend(self.get_raw_css())
-        self.start_periodic_updates()
+        self._start_periodic_callback(session_updater)
         return template
 
     def get_raw_css(self) -> list[str]:

src/ess/livedata/dashboard/dashboard_services.py

@@ -2,9 +2,9 @@
 # Copyright (c) 2025 Scipp contributors (https://github.com/scipp)
 """Dashboard service composition and setup."""
 
-from collections.abc import Callable
+import threading
+import time
 from contextlib import ExitStack
-from typing import Any
 
 import scipp as sc
 import structlog
@@ -20,11 +20,14 @@
 from .job_controller import JobController
 from .job_orchestrator import JobOrchestrator
 from .job_service import JobService
+from .notification_queue import NotificationQueue
 from .orchestrator import Orchestrator
+from .plot_data_service import PlotDataService
 from .plot_orchestrator import PlotOrchestrator
 from .plotting_controller import PlottingController
 from .roi_publisher import ROIPublisher
 from .service_registry import ServiceRegistry
+from .session_registry import SessionRegistry
 from .stream_manager import StreamManager
 from .transport import Transport
 from .workflow_controller import WorkflowController
@@ -49,10 +52,6 @@ class DashboardServices:
         Use dev mode with simplified topic structure
     exit_stack:
         ExitStack for managing resource cleanup (caller manages lifecycle)
-    pipe_factory:
-        Factory function for creating pipes for StreamManager.
-        For GUI: use holoviews.streams.Pipe
-        For tests: use lambda data: None (no-op)
     transport:
         Transport instance for message sources and sinks.
         For Kafka: DashboardKafkaTransport(instrument, dev)
@@ -70,21 +69,31 @@ def __init__(
         instrument: str,
         dev: bool,
         exit_stack: ExitStack,
-        pipe_factory: Callable[[Any], Any],
         transport: Transport,
         config_manager: ConfigStoreManager,
     ):
         self._instrument = instrument
         self._dev = dev
         self._exit_stack = exit_stack
-        self._pipe_factory = pipe_factory
         self._transport = transport
         self._config_manager = config_manager
 
         # Config stores for workflow and plotter persistent UI state
         self.workflow_config_store = config_manager.get_store('workflow_configs')
         self.plotter_config_store = config_manager.get_store('plotter_configs')
 
+        # Shared state stores for multi-session support
+        self.plot_data_service = PlotDataService()
+        self.notification_queue = NotificationQueue()
+
+        # Session registry for tracking active browser sessions
+        self.session_registry = SessionRegistry(stale_timeout_seconds=60.0)
+
+        # Background update thread state
+        self._update_thread: threading.Thread | None = None
+        self._stop_event = threading.Event()
+        self._update_interval = 0.2  # seconds
+
         # Setup all services
         self._setup_data_infrastructure()
         self._setup_workflow_management()
@@ -93,13 +102,53 @@ def __init__(
         logger.info("DashboardServices initialized for %s", instrument)
 
     def start(self) -> None:
-        """Start background tasks (e.g., message polling)."""
+        """Start background tasks (message polling and orchestrator updates)."""
         self._transport.start()
+        self._start_update_thread()
 
     def stop(self) -> None:
-        """Stop background tasks (e.g., message polling)."""
+        """Stop background tasks (message polling and orchestrator updates)."""
+        self._stop_update_thread()
         self._transport.stop()
 
+    def _start_update_thread(self) -> None:
+        """Start the background thread that runs orchestrator updates."""
+        if self._update_thread is not None:
+            return
+        self._stop_event.clear()
+        self._update_thread = threading.Thread(
+            target=self._update_loop, daemon=True, name="orchestrator-update"
+        )
+        self._update_thread.start()
+        logger.info("orchestrator_update_thread_started")
+
+    def _stop_update_thread(self) -> None:
+        """Stop the background orchestrator update thread."""
+        if self._update_thread is None:
+            return
+        self._stop_event.set()
+        self._update_thread.join(timeout=5.0)
+        if self._update_thread.is_alive():
+            logger.warning("orchestrator_update_thread_stop_timeout")
+        self._update_thread = None
+        logger.info("orchestrator_update_thread_stopped")
+
+    def _update_loop(self) -> None:
+        """Background loop that periodically updates the orchestrator."""
+        while not self._stop_event.is_set():
+            start = time.monotonic()
+            try:
+                self.orchestrator.update()
+                self.session_registry.cleanup_stale_sessions()
+            except Exception:
+                logger.exception("orchestrator_update_error")
+
+            # Only sleep if update was quick (no work to do), to avoid busy-wait.
+            # If there was real work, loop immediately to check for more data.
+            elapsed = time.monotonic() - start
+            if elapsed < 0.01:  # < 10ms means no real work
+                self._stop_event.wait(self._update_interval)
+
     def _setup_data_infrastructure(self) -> None:
         """Set up data services, forwarder, and orchestrator."""
         # Set up transport and get resources
@@ -110,9 +159,7 @@ def _setup_data_infrastructure(self) -> None:
         # da00 of backend services converted to scipp.DataArray
         ScippDataService = DataService[ResultKey, sc.DataArray]
         self.data_service = ScippDataService()
-        self.stream_manager = StreamManager(
-            data_service=self.data_service, pipe_factory=self._pipe_factory
-        )
+        self.stream_manager = StreamManager(data_service=self.data_service)
         self.job_service = JobService()
         self.service_registry = ServiceRegistry()
         self.job_controller = JobController(
@@ -145,6 +192,7 @@ def _setup_plot_orchestrator(self) -> None:
             config_store=plot_config_store,
             raw_templates=raw_templates,
             instrument_config=self.instrument_config,
+            plot_data_service=self.plot_data_service,
         )
         logger.info("PlotOrchestrator setup complete")
 
@@ -160,6 +208,7 @@ def _setup_workflow_management(self) -> None:
             workflow_registry=self.processor_factory,
             config_store=self.workflow_config_store,
             instrument_config=self.instrument_config,
+            notification_queue=self.notification_queue,
         )
         self.workflow_controller = WorkflowController(
             job_orchestrator=self.job_orchestrator,

src/ess/livedata/dashboard/data_service.py

@@ -164,6 +164,26 @@ def register_subscriber(self, subscriber: DataServiceSubscriber[K]) -> None:
         existing_data = self._build_subscriber_data(subscriber)
         subscriber.trigger(existing_data)
 
+    def unregister_subscriber(self, subscriber: DataServiceSubscriber[K]) -> bool:
+        """
+        Unregister a subscriber, stopping it from receiving updates.
+
+        Parameters
+        ----------
+        subscriber:
+            The subscriber to unregister.
+
+        Returns
+        -------
+        :
+            True if the subscriber was found and removed, False otherwise.
+        """
+        try:
+            self._subscribers.remove(subscriber)
+            return True
+        except ValueError:
+            return False
+
     def _notify_subscribers(self, updated_keys: set[K]) -> None:
         """
         Notify relevant subscribers about data updates.