refactor: deduplicate streaming methods across all modalities

src/celeste/modalities/audio/providers/elevenlabs/client.py

@@ -26,47 +26,14 @@
 class ElevenLabsAudioStream(_ElevenLabsTextToSpeechStream, AudioStream):
     """ElevenLabs streaming for audio modality."""
 
-    def _parse_chunk(self, event_data: dict[str, Any]) -> AudioChunk | None:
-        """Parse binary audio chunk from stream event."""
-        chunk_data = self._parse_chunk_content(event_data)
-        if not chunk_data:
-            usage = self._get_chunk_usage(event_data)
-            finish_reason = self._get_chunk_finish_reason(event_data)
-            if usage is None and finish_reason is None:
-                return None
-            # Chunk with usage/finish_reason only (no audio)
-            return AudioChunk(
-                content=b"",
-                finish_reason=finish_reason,
-                usage=usage,
-                metadata={"event_data": event_data},
-            )
-
-        return AudioChunk(
-            content=chunk_data,
-            finish_reason=self._get_chunk_finish_reason(event_data),
-            usage=self._get_chunk_usage(event_data),
-            metadata={"event_data": event_data},
-        )
-
     def _aggregate_content(self, chunks: list[AudioChunk]) -> AudioArtifact:
         """Aggregate audio content from chunks into AudioArtifact."""
         audio_bytes = b"".join(chunk.content for chunk in chunks if chunk.content)
-        # Get mime_type from output_format parameter via client
         output_format = self._parameters.get("output_format")
         client: ElevenLabsAudioClient = self._client
         mime_type = client._map_output_format_to_mime_type(output_format)
         return AudioArtifact(data=audio_bytes, mime_type=mime_type)
 
-    def _aggregate_event_data(self, chunks: list[AudioChunk]) -> list[dict[str, Any]]:
-        """Collect raw events (filtering happens in _build_stream_metadata)."""
-        events: list[dict[str, Any]] = []
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
-        return events
-
 
 class ElevenLabsAudioClient(ElevenLabsTextToSpeechMixin, AudioClient):
     """ElevenLabs audio client (TTS)."""

src/celeste/modalities/audio/providers/gradium/client.py

@@ -26,47 +26,14 @@
 class GradiumAudioStream(_GradiumTextToSpeechStream, AudioStream):
     """Gradium streaming for audio modality."""
 
-    def _parse_chunk(self, event_data: dict[str, Any]) -> AudioChunk | None:
-        """Parse binary audio chunk from stream event."""
-        chunk_data = self._parse_chunk_content(event_data)
-        if not chunk_data:
-            usage = self._get_chunk_usage(event_data)
-            finish_reason = self._get_chunk_finish_reason(event_data)
-            if usage is None and finish_reason is None:
-                return None
-            # Chunk with usage/finish_reason only (no audio)
-            return AudioChunk(
-                content=b"",
-                finish_reason=finish_reason,
-                usage=usage,
-                metadata={"event_data": event_data},
-            )
-
-        return AudioChunk(
-            content=chunk_data,
-            finish_reason=self._get_chunk_finish_reason(event_data),
-            usage=self._get_chunk_usage(event_data),
-            metadata={"event_data": event_data},
-        )
-
     def _aggregate_content(self, chunks: list[AudioChunk]) -> AudioArtifact:
         """Aggregate audio content from chunks into AudioArtifact."""
         audio_bytes = b"".join(chunk.content for chunk in chunks if chunk.content)
-        # Get mime_type from output_format parameter via client
         output_format = self._parameters.get("output_format")
         client: GradiumAudioClient = self._client
         mime_type = client._map_output_format_to_mime_type(output_format)
         return AudioArtifact(data=audio_bytes, mime_type=mime_type)
 
-    def _aggregate_event_data(self, chunks: list[AudioChunk]) -> list[dict[str, Any]]:
-        """Collect raw events (filtering happens in _build_stream_metadata)."""
-        events: list[dict[str, Any]] = []
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
-        return events
-
 
 class GradiumAudioClient(GradiumTextToSpeechMixin, AudioClient):
     """Gradium audio client (TTS)."""

src/celeste/modalities/audio/streaming.py

@@ -22,6 +22,8 @@ class AudioStream(Stream[AudioOutput, AudioParameters, AudioChunk]):
 
     _usage_class = AudioUsage
     _finish_reason_class = AudioFinishReason
+    _chunk_class = AudioChunk
+    _empty_content = b""
 
     def __init__(
         self,
@@ -39,28 +41,6 @@ def _aggregate_content(self, chunks: list[AudioChunk]) -> AudioArtifact:
         """Aggregate content from chunks into AudioArtifact."""
         ...
 
-    def _aggregate_usage(self, chunks: list[AudioChunk]) -> AudioUsage:
-        """Aggregate usage across chunks."""
-        for chunk in reversed(chunks):
-            if chunk.usage:
-                return chunk.usage
-        return AudioUsage()
-
-    def _aggregate_finish_reason(
-        self,
-        chunks: list[AudioChunk],
-    ) -> AudioFinishReason | None:
-        """Aggregate finish reason across chunks."""
-        for chunk in reversed(chunks):
-            if chunk.finish_reason:
-                return chunk.finish_reason
-        return None
-
-    @abstractmethod
-    def _aggregate_event_data(self, chunks: list[AudioChunk]) -> list[dict[str, Any]]:
-        """Collect raw events."""
-        ...
-
     def _build_stream_metadata(
         self, raw_events: list[dict[str, Any]]
     ) -> dict[str, Any]:

src/celeste/modalities/images/providers/byteplus/client.py

@@ -82,14 +82,11 @@ def _aggregate_usage(self, chunks: list[ImageChunk]) -> ImageUsage:
         return super()._aggregate_usage(chunks)
 
     def _aggregate_event_data(self, chunks: list[ImageChunk]) -> list[dict[str, Any]]:
-        """Collect metadata events (skip content-only events)."""
+        """Prepend completed_event_data, then delegate to base."""
         events: list[dict[str, Any]] = []
         if self._completed_event_data is not None:
             events.append(self._completed_event_data)
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
+        events.extend(super()._aggregate_event_data(chunks))
         return events
 
 

src/celeste/modalities/images/providers/ollama/client.py

@@ -32,14 +32,14 @@ def _parse_chunk(self, event_data: dict[str, Any]) -> ImageChunk | None:
         if not b64_image:
             return ImageChunk(
                 content=ImageArtifact(data=b""),
-                metadata=self._parse_chunk_metadata(event_data),
+                metadata={"event_data": event_data},
             )
 
         return ImageChunk(
             content=ImageArtifact(data=b64_image),
             finish_reason=self._get_chunk_finish_reason(event_data),
             usage=self._get_chunk_usage(event_data),
-            metadata=self._parse_chunk_metadata(event_data),
+            metadata={"event_data": event_data},
         )
 
     def _aggregate_content(self, chunks: list[ImageChunk]) -> ImageArtifact:
@@ -50,10 +50,6 @@ def _aggregate_content(self, chunks: list[ImageChunk]) -> ImageArtifact:
         msg = "No image in stream"
         raise ValueError(msg)
 
-    def _aggregate_event_data(self, chunks: list[ImageChunk]) -> list[dict[str, Any]]:
-        """Collect metadata from chunks."""
-        return [chunk.metadata for chunk in chunks if chunk.metadata]
-
 
 class OllamaImagesClient(OllamaGenerateClient, ImagesClient):
     """Ollama images client (generate only, no edit support yet)."""

src/celeste/modalities/images/providers/openai/client.py

@@ -26,44 +26,10 @@
 class OpenAIImagesStream(_OpenAIImagesStream, ImagesStream):
     """OpenAI streaming for images modality."""
 
-    def _parse_chunk(self, event_data: dict[str, Any]) -> ImageChunk | None:
-        """Parse one SSE event into a typed chunk."""
-        b64_json = self._parse_chunk_content(event_data)
-        if not b64_json:
-            usage = self._get_chunk_usage(event_data)
-            finish_reason = self._get_chunk_finish_reason(event_data)
-            if usage is None and finish_reason is None:
-                return None
-            # Chunk with usage/finish_reason only (no image)
-            return ImageChunk(
-                content=ImageArtifact(data=b""),
-                finish_reason=finish_reason,
-                usage=usage,
-                metadata={"event_data": event_data},
-            )
-
-        artifact = ImageArtifact(data=b64_json)
-
-        return ImageChunk(
-            content=artifact,
-            finish_reason=self._get_chunk_finish_reason(event_data),
-            usage=self._get_chunk_usage(event_data),
-            metadata={"event_data": event_data},
-        )
-
     def _aggregate_content(self, chunks: list[ImageChunk]) -> ImageArtifact:
         """Aggregate image content from chunks."""
         return chunks[-1].content
 
-    def _aggregate_event_data(self, chunks: list[ImageChunk]) -> list[dict[str, Any]]:
-        """Collect metadata events (skip content-only events)."""
-        events: list[dict[str, Any]] = []
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
-        return events
-
 
 class OpenAIImagesClient(OpenAIImagesMixin, ImagesClient):
     """OpenAI images client."""

src/celeste/modalities/images/streaming.py

@@ -18,6 +18,8 @@ class ImagesStream(Stream[ImageOutput, ImageParameters, ImageChunk]):
 
     _usage_class = ImageUsage
     _finish_reason_class = ImageFinishReason
+    _chunk_class = ImageChunk
+    _empty_content = ImageArtifact(data=b"")
 
     def __init__(
         self,
@@ -31,33 +33,15 @@ def __init__(
         self._transform_output = transform_output
         self._client = client
 
+    def _wrap_chunk_content(self, raw_content: Any) -> ImageArtifact:  # noqa: ANN401
+        """Wrap raw content (base64 string) into ImageArtifact."""
+        return ImageArtifact(data=raw_content)
+
     @abstractmethod
     def _aggregate_content(self, chunks: list[ImageChunk]) -> ImageArtifact:
         """Aggregate content from chunks into raw content (modality-specific)."""
         ...
 
-    def _aggregate_usage(self, chunks: list[ImageChunk]) -> ImageUsage:
-        """Aggregate usage across chunks (universal)."""
-        for chunk in reversed(chunks):
-            if chunk.usage:
-                return chunk.usage
-        return ImageUsage()
-
-    def _aggregate_finish_reason(
-        self,
-        chunks: list[ImageChunk],
-    ) -> ImageFinishReason | None:
-        """Aggregate finish reason across chunks (universal)."""
-        for chunk in reversed(chunks):
-            if chunk.finish_reason:
-                return chunk.finish_reason
-        return None
-
-    @abstractmethod
-    def _aggregate_event_data(self, chunks: list[ImageChunk]) -> list[dict[str, Any]]:
-        """Collect raw events (filtering happens in _build_stream_metadata)."""
-        ...
-
     def _build_stream_metadata(
         self, raw_events: list[dict[str, Any]]
     ) -> dict[str, Any]:

src/celeste/modalities/text/providers/anthropic/client.py

@@ -32,42 +32,21 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         self._message_start: dict[str, Any] | None = None
 
     def _parse_chunk(self, event_data: dict[str, Any]) -> TextChunk | None:
-        """Parse one SSE event into a typed chunk."""
+        """Parse one SSE event into a typed chunk (captures message_start)."""
         event_type = event_data.get("type")
         if event_type == "message_start":
             message = event_data.get("message")
             if isinstance(message, dict):
                 self._message_start = message
             return None
-
-        content = self._parse_chunk_content(event_data)
-        if content is None:
-            usage = self._get_chunk_usage(event_data)
-            finish_reason = self._get_chunk_finish_reason(event_data)
-            if usage is None and finish_reason is None:
-                return None
-            content = ""
-
-        return TextChunk(
-            content=content,
-            finish_reason=self._get_chunk_finish_reason(event_data),
-            usage=self._get_chunk_usage(event_data),
-            metadata={"event_data": event_data},
-        )
-
-    def _aggregate_content(self, chunks: list[TextChunk]) -> str:
-        """Aggregate streamed text content."""
-        return "".join(chunk.content for chunk in chunks)
+        return super()._parse_chunk(event_data)
 
     def _aggregate_event_data(self, chunks: list[TextChunk]) -> list[dict[str, Any]]:
-        """Collect raw events (filtering happens in _build_stream_metadata)."""
+        """Prepend message_start, then delegate to base."""
         events: list[dict[str, Any]] = []
         if self._message_start is not None:
             events.append({"type": "message_start", "message": self._message_start})
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
+        events.extend(super()._aggregate_event_data(chunks))
         return events
 
 

src/celeste/modalities/text/providers/cohere/client.py

@@ -12,7 +12,6 @@
 
 from ...client import TextClient
 from ...io import (
-    TextChunk,
     TextInput,
     TextOutput,
 )
@@ -24,36 +23,6 @@
 class CohereTextStream(_CohereChatStream, TextStream):
     """Cohere streaming for text modality."""
 
-    def _parse_chunk(self, event_data: dict[str, Any]) -> TextChunk | None:
-        """Parse one SSE event into a typed chunk."""
-        content = self._parse_chunk_content(event_data)
-        if content is None:
-            usage = self._get_chunk_usage(event_data)
-            finish_reason = self._get_chunk_finish_reason(event_data)
-            if usage is None and finish_reason is None:
-                return None
-            content = ""
-
-        return TextChunk(
-            content=content,
-            finish_reason=self._get_chunk_finish_reason(event_data),
-            usage=self._get_chunk_usage(event_data),
-            metadata={"event_data": event_data},
-        )
-
-    def _aggregate_content(self, chunks: list[TextChunk]) -> str:
-        """Aggregate streamed text content."""
-        return "".join(chunk.content for chunk in chunks)
-
-    def _aggregate_event_data(self, chunks: list[TextChunk]) -> list[dict[str, Any]]:
-        """Collect raw events (filtering happens in _build_stream_metadata)."""
-        events: list[dict[str, Any]] = []
-        for chunk in chunks:
-            event_data = chunk.metadata.get("event_data")
-            if isinstance(event_data, dict):
-                events.append(event_data)
-        return events
-
 
 class CohereTextClient(CohereChatClient, TextClient):
     """Cohere text client."""