Extract receiver-core projection/batching primitives and unify Fiji/Napari batch paths

This change moves streaming receiver mechanics into shared core abstractions and
rewires Fiji/Napari receiver implementations to consume those primitives.

Highlights:
- Add receiver-core package:
  - batching contract ABC
  - debounced batch engine
  - window/component projection contracts and helpers
- Refactor streaming base interfaces to explicit ABC boundaries (no protocol/fallback path).
- Factor generic layer-key/component-layout helpers for napari receivers.
- Refactor Fiji and Napari batch processors to use DebouncedBatchEngine and shared
  projection semantics instead of duplicated local debounce/state code.
- Add architecture docs for streaming receiver projection boundaries.
- Add receiver-core tests, including non-blocking enqueue behavior while processing.

Outcome:
- Single reusable batching/projection path for receiver implementations
- reduced duplicate debounce logic and clearer separation of concerns
- stronger extractability for downstream streaming servers.

Author: trissim

docs/source/architecture/index.rst

@@ -0,0 +1,9 @@
+Architecture
+============
+
+High-level architecture for polystore's storage and streaming abstractions.
+
+.. toctree::
+   :maxdepth: 1
+
+   streaming_receiver_projection

docs/source/architecture/streaming_receiver_projection.rst

@@ -0,0 +1,65 @@
+Streaming Receiver Projection
+=============================
+
+Modules
+-------
+
+- ``polystore.streaming.receivers.core.contracts``
+- ``polystore.streaming.receivers.core.batch_engine``
+- ``polystore.streaming.receivers.core.window_projection``
+- ``polystore.streaming.receivers.napari.layer_key``
+
+Purpose
+-------
+
+Provide reusable, viewer-agnostic receiver-side primitives for streaming
+payload projection and batched update scheduling.
+
+Boundary
+--------
+
+``polystore`` owns payload semantics and receiver projection mechanics:
+
+- component-mode grouping into window/channel/slice/frame structures
+- canonical layer-key derivation from component metadata
+- debounced batch processing with bounded wait behavior
+- receiver contracts via nominal ABCs
+
+``polystore`` does not own ZMQ transport lifecycle. Transport/server ownership
+belongs to ``zmqruntime``.
+
+Core Contracts
+--------------
+
+``BatchEngineABC``
+  Contract for enqueue/flush behavior in receiver-side batch schedulers.
+
+``WindowProjectionABC``
+  Contract for grouping stream items into projected window structures.
+
+Reference Implementations
+-------------------------
+
+``DebouncedBatchEngine``
+  Thread-safe debounce + max-wait engine for coalescing incoming items before
+  projection/render updates.
+
+``group_items_by_component_modes``
+  Canonical grouping utility that projects incoming items by declared component
+  modes and returns stable ``GroupedWindowItems`` output.
+
+``build_layer_key``
+  Canonical napari layer-key construction from slice-mode components and data
+  type.
+
+Design Outcome
+--------------
+
+Receiver implementations (for example napari and Fiji wrappers in downstream
+applications) can share one projection/batching kernel while keeping
+viewer-specific rendering code separate.
+
+See Also
+--------
+
+- ``external/zmqruntime/docs/source/architecture/viewer_streaming_architecture.rst``

docs/source/index.rst

@@ -11,6 +11,7 @@ Polystore provides a pluggable storage backend system with multi-framework I/O s
 
    installation
    quickstart
+   architecture/index
    api/index
    guides/custom_backends
 

src/polystore/streaming/base.py

@@ -5,7 +5,8 @@
 that work with arbitrary numbers of components (not hardcoded to 3).
 """
 
-from typing import TypeVar, Generic, Dict, Any, Protocol, runtime_checkable
+from abc import ABC, abstractmethod
+from typing import TypeVar, Generic, Dict, Any, List
 from dataclasses import dataclass
 
 T = TypeVar('T')
@@ -29,9 +30,10 @@ class TypedData(Generic[T]):
     source: str
 
 
-class ComponentAccessor(Protocol):
-    """Protocol for component metadata access (arbitrary number of components)."""
+class ComponentAccessor(ABC):
+    """ABC for component metadata access (arbitrary number of components)."""
 
+    @abstractmethod
     def get_by_mode(self, mode: str) -> list:
         """
         Get all component names that have this mode (stack/slice/window).
@@ -46,42 +48,68 @@ def get_by_mode(self, mode: str) -> list:
             If config has {'channel': 'stack', 'z_index': 'slice', 'well': 'window'}
             Then get_by_mode('stack') returns ['channel']
         """
-        ...
+        raise NotImplementedError
 
+    @abstractmethod
     def get_value(self, item: Dict[str, Any], component_name: str) -> Any:
         """
         Get component value for an item.
 
         Returns:
             Value or default (0) if component not in metadata.
         """
-        ...
+        raise NotImplementedError
 
+    @abstractmethod
     def collect_values(self, component_names: list) -> list[tuple]:
         """
         Collect unique values for given components across all items.
 
         Returns:
             Sorted list of tuples for consistent indexing.
         """
-        ...
+        raise NotImplementedError
 
 
-class HandlerContext(Protocol):
-    """Protocol for handler context with generic component access."""
+class HandlerContext(ABC):
+    """ABC for handler context with generic component access."""
 
-    server: Any
-    window_key: str
-    data: 'TypedData[Any]'
-    display_config: Dict[str, Any]
-    components: ComponentAccessor
-    images_dir: str | None
+    @property
+    @abstractmethod
+    def server(self) -> Any:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def window_key(self) -> str:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def data(self) -> "TypedData[Any]":
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def display_config(self) -> Dict[str, Any]:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def components(self) -> ComponentAccessor:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def images_dir(self) -> str | None:
+        raise NotImplementedError
 
 
-class ItemHandler(Protocol):
-    """Type-safe protocol for item handlers with automatic discovery."""
+class ItemHandler(ABC):
+    """Type-safe ABC for item handlers with automatic discovery."""
 
     @staticmethod
+    @abstractmethod
     def can_handle(data_type: str) -> bool:
         """
         Check if this handler can process the given data type.
@@ -92,17 +120,18 @@ def can_handle(data_type: str) -> bool:
         Returns:
             True if this handler can process this type.
         """
-        ...
+        raise NotImplementedError
 
     @staticmethod
+    @abstractmethod
     def handle(context: HandlerContext) -> None:
         """
         Process items using type-safe context object.
 
         Args:
             context: HandlerContext with typed data and component accessor.
         """
-        ...
+        raise NotImplementedError
 
 
 @dataclass(frozen=True)
@@ -171,7 +200,7 @@ def collect_values(self, component_names: list) -> list[tuple]:
 
 
 @dataclass(frozen=True)
-class SimpleHandlerContext(HandlerContext):
+class SimpleHandlerContext:
     """Concrete implementation of HandlerContext protocol."""
 
     server: Any

src/polystore/streaming/handlers/fiji_images.py

@@ -40,10 +40,7 @@ def can_handle(data_type: str) -> bool:
     @staticmethod
     def handle(context: HandlerContext) -> None:
         """Build hyperstack from accumulated images."""
-        # Access data via typed wrapper
-        images = context.data
-
-        # Access components generically (no hardcoded 3 dimensions!)
+        images = context.data.items
         channel_comps = context.components.get_by_mode("channel")
         slice_comps = context.components.get_by_mode("slice")
         frame_comps = context.components.get_by_mode("frame")
@@ -54,27 +51,20 @@ def handle(context: HandlerContext) -> None:
         frame_values = context.components.collect_values(frame_comps)
 
         logger.info(
-            f"🔬 FIJI IMAGE HANDLER: Processing {len(images.data)} images: "
+            f"🔬 FIJI IMAGE HANDLER: Processing {len(images)} images: "
             f"{len(channel_values)}C x {len(slice_values)}Z x {len(frame_values)}T"
         )
-
-        # Build hyperstack using server's method
-        # Note: We don't call _build_single_hyperstack directly - we need to adapt
-        # the existing server code to work with our new architecture.
-        # For now, this is a placeholder that logs the handler was called.
-        # In a full implementation, this would call the actual hyperstack building
-        # logic from FijiViewerServer.
-
-        # The actual implementation would be:
-        # context.server._build_single_hyperstack(
-        #     window_key=context.window_key,
-        #     images=images.data,
-        #     display_config_dict=context.display_config,
-        #     channel_components=channel_comps,
-        #     slice_components=slice_comps,
-        #     frame_components=frame_comps,
-        #     channel_values=channel_values,
-        #     slice_values=slice_values,
-        #     frame_values=frame_values,
-        #     component_names_metadata=context.display_config.get('component_names_metadata', {}),
-        # )
+        context.server._build_single_hyperstack(
+            window_key=context.window_key,
+            images=images,
+            display_config_dict=context.display_config,
+            channel_components=channel_comps,
+            slice_components=slice_comps,
+            frame_components=frame_comps,
+            channel_values=channel_values,
+            slice_values=slice_values,
+            frame_values=frame_values,
+            component_names_metadata=context.display_config.get(
+                "component_names_metadata", {}
+            ),
+        )

src/polystore/streaming/receivers/__init__.py

@@ -4,11 +4,28 @@
 Provides reusable batching and debouncing logic for Fiji and Napari viewers.
 """
 
+from polystore.streaming.receivers.core import (
+    BatchEngineABC,
+    WindowProjectionABC,
+    DebouncedBatchEngine,
+    GroupedWindowItems,
+    group_items_by_component_modes,
+)
 from polystore.streaming.receivers.fiji.fiji_batch_processor import FijiBatchProcessor
-from polystore.streaming.receivers.napari.napari_batch_processor import NapariBatchProcessor
+from polystore.streaming.receivers.napari import (
+    NapariBatchProcessor,
+    normalize_component_layout,
+    build_layer_key,
+)
 
 __all__ = [
+    "BatchEngineABC",
+    "WindowProjectionABC",
+    "DebouncedBatchEngine",
+    "GroupedWindowItems",
+    "group_items_by_component_modes",
     "FijiBatchProcessor",
     "NapariBatchProcessor",
+    "normalize_component_layout",
+    "build_layer_key",
 ]
-

src/polystore/streaming/receivers/core/__init__.py

@@ -0,0 +1,20 @@
+"""Shared receiver-core utilities for streaming viewers."""
+
+from polystore.streaming.receivers.core.batch_engine import DebouncedBatchEngine
+from polystore.streaming.receivers.core.contracts import (
+    BatchEngineABC,
+    WindowProjectionABC,
+)
+from polystore.streaming.receivers.core.window_projection import (
+    GroupedWindowItems,
+    group_items_by_component_modes,
+)
+
+__all__ = [
+    "BatchEngineABC",
+    "WindowProjectionABC",
+    "DebouncedBatchEngine",
+    "GroupedWindowItems",
+    "group_items_by_component_modes",
+]
+