Processors

The two stream processors and their configuration. See the architecture overview for how they fit in the pipeline.

hother.streamblocks.core.processor

Stream processing engine for StreamBlocks.

TChunk module-attribute

TChunk = TypeVar('TChunk')

ProcessorConfig dataclass

Configuration for StreamBlockProcessor.

Attributes:

Name	Type	Description
lines_buffer	int	Number of recent lines to keep in buffer for context (default: 5). Used for debugging and error messages.
max_line_length	int	Maximum line length in bytes before truncation (default: 16,384). Lines exceeding this limit are truncated to prevent memory issues.
max_block_size	int	Maximum block size in bytes before rejection (default: 1,048,576 = 1MB). Blocks exceeding this limit are rejected with SIZE_EXCEEDED error.
emit_original_events	bool	Whether to pass through original provider events (default: True). When False, only StreamBlocks events are emitted. Set to False when using IdentityInputAdapter to avoid duplicate events.
emit_text_deltas	bool	Whether to emit TextDeltaEvent for real-time streaming (default: True). Enables character-level streaming for live UIs. Disable to reduce event volume.
emit_section_end_events	bool	Whether to emit section end events (default: True). Controls BlockMetadataEndEvent and BlockContentEndEvent emission for early validation.
auto_detect_adapter	bool	Whether to auto-detect input adapter from first chunk (default: True). When False, uses IdentityInputAdapter. Disable for performance with known adapter.

Example

Custom configuration for large blocks

config = ProcessorConfig( ... max_block_size=2_097_152, # 2MB ... emit_original_events=False, ... emit_text_deltas=False, ... ) processor = StreamBlockProcessor(registry, config=config)

Minimal configuration for performance

config = ProcessorConfig( ... emit_section_end_events=False, ... auto_detect_adapter=False, ... )

Source code in src/hother/streamblocks/core/processor.py

@dataclass(frozen=True, slots=True)
class ProcessorConfig:
    """Configuration for StreamBlockProcessor.

    Attributes:
        lines_buffer: Number of recent lines to keep in buffer for context (default: 5).
            Used for debugging and error messages.
        max_line_length: Maximum line length in bytes before truncation (default: 16,384).
            Lines exceeding this limit are truncated to prevent memory issues.
        max_block_size: Maximum block size in bytes before rejection (default: 1,048,576 = 1MB).
            Blocks exceeding this limit are rejected with SIZE_EXCEEDED error.
        emit_original_events: Whether to pass through original provider events (default: True).
            When False, only StreamBlocks events are emitted. Set to False when using
            IdentityInputAdapter to avoid duplicate events.
        emit_text_deltas: Whether to emit TextDeltaEvent for real-time streaming (default: True).
            Enables character-level streaming for live UIs. Disable to reduce event volume.
        emit_section_end_events: Whether to emit section end events (default: True).
            Controls BlockMetadataEndEvent and BlockContentEndEvent emission for early validation.
        auto_detect_adapter: Whether to auto-detect input adapter from first chunk (default: True).
            When False, uses IdentityInputAdapter. Disable for performance with known adapter.

    Example:
        >>> # Custom configuration for large blocks
        >>> config = ProcessorConfig(
        ...     max_block_size=2_097_152,  # 2MB
        ...     emit_original_events=False,
        ...     emit_text_deltas=False,
        ... )
        >>> processor = StreamBlockProcessor(registry, config=config)
        >>>
        >>> # Minimal configuration for performance
        >>> config = ProcessorConfig(
        ...     emit_section_end_events=False,
        ...     auto_detect_adapter=False,
        ... )
    """

    lines_buffer: int = LIMITS.LINES_BUFFER
    max_line_length: int = LIMITS.MAX_LINE_LENGTH
    max_block_size: int = LIMITS.MAX_BLOCK_SIZE
    emit_original_events: bool = True
    emit_text_deltas: bool = True
    emit_section_end_events: bool = True
    auto_detect_adapter: bool = True

auto_detect_adapter class-attribute instance-attribute

auto_detect_adapter: bool = True

emit_original_events class-attribute instance-attribute

emit_original_events: bool = True

emit_section_end_events class-attribute instance-attribute

emit_section_end_events: bool = True

emit_text_deltas class-attribute instance-attribute

emit_text_deltas: bool = True

lines_buffer class-attribute instance-attribute

lines_buffer: int = LINES_BUFFER

max_block_size class-attribute instance-attribute

max_block_size: int = MAX_BLOCK_SIZE

max_line_length class-attribute instance-attribute

max_line_length: int = MAX_LINE_LENGTH

StreamBlockProcessor

Main stream processing engine for a single syntax type.

This processor works with exactly one syntax and coordinates: - Adapter detection and text extraction - Line accumulation via LineAccumulator - Block detection and extraction via BlockStateMachine - Event emission (TextDeltaEvent, block events, etc.)

Source code in src/hother/streamblocks/core/processor.py

class StreamBlockProcessor:
    """Main stream processing engine for a single syntax type.

    This processor works with exactly one syntax and coordinates:
    - Adapter detection and text extraction
    - Line accumulation via LineAccumulator
    - Block detection and extraction via BlockStateMachine
    - Event emission (TextDeltaEvent, block events, etc.)
    """

    def __init__(
        self,
        registry: Registry,
        config: ProcessorConfig | None = None,
        *,
        logger: Logger | None = None,
        state_machine_factory: Callable[..., BlockStateMachine] = BlockStateMachine,
        accumulator_factory: Callable[..., LineAccumulator] = LineAccumulator,
    ) -> None:
        """Initialize the stream processor.

        Args:
            registry: Registry with a single syntax
            config: Configuration object for processor settings
            logger: Optional logger (any object with debug/info/warning/error/exception methods).
                   Defaults to stdlib logging.getLogger(__name__)
            state_machine_factory: Factory for creating BlockStateMachine (dependency injection)
            accumulator_factory: Factory for creating LineAccumulator (dependency injection)
        """
        self.registry = registry
        self.syntax = registry.syntax
        self.logger = logger or StdlibLoggerAdapter(logging.getLogger(__name__))
        self.config = config or ProcessorConfig()

        # Processing components
        self._line_accumulator = accumulator_factory(
            max_line_length=self.config.max_line_length,
            buffer_size=self.config.lines_buffer,
        )
        self._block_machine = state_machine_factory(
            syntax=self.syntax,
            registry=registry,
            max_block_size=self.config.max_block_size,
            emit_section_end_events=self.config.emit_section_end_events,
            logger=self.logger,
        )
        self._stream_state = StreamState()

        # Adapter state
        self._adapter: InputProtocolAdapter[Any] | None = None
        self._first_chunk_processed = False

    def process_chunk(
        self,
        chunk: TChunk,
        adapter: InputProtocolAdapter[TChunk] | None = None,
    ) -> list[TChunk | Event]:
        """Process a single chunk and return resulting events.

        This method is stateful - it maintains internal state between calls.
        Call finalize() after processing all chunks to flush incomplete blocks.

        Args:
            chunk: Single chunk to process
            adapter: Optional adapter for extracting text. If not provided and
                    auto_detect_adapter=True, will auto-detect on first chunk.

        Returns:
            List of events generated from this chunk. May be empty if chunk only
            accumulates text without completing any lines.

        Raises:
            RuntimeError: If adapter is not set after first chunk processing
                (internal state error, should not occur in normal usage).

        Example:
            >>> processor = StreamBlockProcessor(registry)
            >>> response = await client.generate_content_stream(...)
            >>> async for chunk in response:
            ...     events = processor.process_chunk(chunk)
            ...     for event in events:
            ...         if isinstance(event, BlockEndEvent):
            ...             print(f"Block: {event.block_id}")
            ...
            >>> # Finalize at stream end
            >>> final_events = processor.finalize()
            >>> for event in final_events:
            ...     if isinstance(event, BlockErrorEvent):
            ...         print(f"Incomplete block: {event.reason}")
        """
        return list(self._iter_chunk_outputs(chunk, adapter, context="process_chunk"))

    def finalize(self) -> list[Event]:
        """Finalize processing and flush any incomplete blocks.

        Call this method after processing all chunks to get rejection events
        for any blocks that were opened but never closed.

        This method processes any accumulated text as a final line before
        flushing candidates, ensuring the last line is processed even if it
        doesn't end with a newline.

        Returns:
            List of events including processed final line and rejection events
            for incomplete blocks

        Example:
            >>> processor = StreamBlockProcessor(registry)
            >>> async for chunk in stream:
            ...     events = processor.process_chunk(chunk)
            ...     # ... handle events
            ...
            >>> # Stream ended, process remaining text and flush incomplete blocks
            >>> final_events = processor.finalize()
            >>> for event in final_events:
            ...     if isinstance(event, BlockErrorEvent):
            ...         print(f"Incomplete block: {event.reason}")
        """
        return list(self._iter_finalize_outputs())

    def is_native_event(self, event: Any) -> bool:
        """Check if event is a native provider event (not a StreamBlocks event).

        This method provides provider-agnostic detection of native events.
        It checks if the event originates from the AI provider (Gemini, OpenAI,
        Anthropic, etc.) versus being a StreamBlocks event.

        Args:
            event: Event to check

        Returns:
            True if event is from the native provider, False if it's a StreamBlocks
            event or if detection is not possible

        Example:
            >>> processor = StreamBlockProcessor(registry)
            >>> async for event in processor.process_stream(gemini_stream):
            ...     if processor.is_native_event(event):
            ...         # Handle Gemini event (provider-agnostic!)
            ...         usage = getattr(event, 'usage_metadata', None)
            ...     elif isinstance(event, BlockEndEvent):
            ...         # Handle StreamBlocks event
            ...         print(f"Block: {event.block_id}")
        """
        # Check if it's a known StreamBlocks event
        if isinstance(
            event,
            (
                StreamStartedEvent,
                StreamFinishedEvent,
                TextContentEvent,
                TextDeltaEvent,
                BlockStartEvent,
                BlockHeaderDeltaEvent,
                BlockMetadataDeltaEvent,
                BlockContentDeltaEvent,
                BlockMetadataEndEvent,
                BlockContentEndEvent,
                BlockEndEvent,
                BlockErrorEvent,
            ),
        ):
            return False

        # Check if we have an adapter with a module prefix
        if self._adapter is None:
            return False

        # Use Protocol-based check for native module prefix
        if not isinstance(self._adapter, HasNativeModulePrefix):
            return False

        # Check if event's module matches the adapter's prefix
        return type(event).__module__.startswith(self._adapter.native_module_prefix)

    async def process_stream(
        self,
        stream: AsyncIterator[TChunk],
        adapter: InputProtocolAdapter[TChunk] | None = None,
    ) -> AsyncGenerator[TChunk | Event]:
        """Process stream and yield mixed events.

        This method processes chunks from any stream format, extracting text
        via an adapter and emitting both original chunks (if enabled) and
        StreamBlocks events.

        Args:
            stream: Async iterator yielding chunks (text or objects)
            adapter: Optional adapter for extracting text from chunks.
                    If None and auto_detect_adapter=True, will auto-detect from first chunk.

        Yields:
            Mixed stream of:
            - Original chunks (if emit_original_events=True)
            - TextDeltaEvent (if emit_text_deltas=True)
            - TextContentEvent, BlockStartEvent, BlockEndEvent, BlockErrorEvent, and section delta events

        Raises:
            RuntimeError: If adapter is not set after first chunk processing
                (internal state error, should not occur in normal usage).

        Example:
            >>> # Plain text
            >>> async for event in processor.process_stream(text_stream):
            ...     if isinstance(event, BlockEndEvent):
            ...         print(f"Extracted: {event.block_id}")
            >>>
            >>> # With Gemini adapter (auto-detected)
            >>> async for event in processor.process_stream(gemini_stream):
            ...     if hasattr(event, 'usage_metadata'):
            ...         print(f"Tokens: {event.usage_metadata}")
            ...     elif isinstance(event, BlockEndEvent):
            ...         print(f"Extracted: {event.block_id}")
        """
        # Set adapter if provided
        if adapter:
            self._adapter = adapter
            self._first_chunk_processed = True

        async for chunk in stream:
            for output in self._iter_chunk_outputs(chunk, None, context="process_stream"):
                yield output

        # Process remaining text and flush incomplete blocks at stream end
        for event in self._iter_finalize_outputs():
            yield event

    def _ensure_adapter(self, chunk: TChunk, adapter: InputProtocolAdapter[TChunk] | None) -> None:
        """Ensure adapter is set, auto-detecting if needed."""
        if self._first_chunk_processed:
            return

        if adapter:
            self._adapter = adapter
        elif self.config.auto_detect_adapter:
            detected = InputAdapterRegistry.detect(chunk)
            if detected:
                self._adapter = detected
                self.logger.info(
                    "adapter_auto_detected",
                    adapter=type(self._adapter).__name__,
                )
            else:
                self._adapter = IdentityInputAdapter()
                self.logger.debug("using_identity_adapter")
        else:
            self._adapter = IdentityInputAdapter()

        self._first_chunk_processed = True

    def _iter_chunk_outputs(
        self,
        chunk: TChunk,
        adapter: InputProtocolAdapter[TChunk] | None,
        *,
        context: str,
    ) -> Iterator[TChunk | Event]:
        """Yield the outputs produced by a single chunk.

        Shared by the sync :meth:`process_chunk` (consumed via ``list()``) and the
        async :meth:`process_stream` (re-yielded). Emits, in order: the original
        chunk (passthrough), an optional TextDeltaEvent, then any block events from
        the lines completed by this chunk's text.

        Args:
            chunk: Single chunk to process.
            adapter: Optional adapter for text extraction; auto-detected on the
                first chunk when not provided.
            context: Caller name used in :class:`AdapterNotConfiguredError`.
        """
        # Auto-detect adapter on first chunk
        self._ensure_adapter(chunk, adapter)

        # Emit original chunk (passthrough)
        if self.config.emit_original_events and not isinstance(self._adapter, IdentityInputAdapter):
            yield chunk

        # Extract text from chunk
        if self._adapter is None:
            raise AdapterNotConfiguredError(context=context)
        text = cast("InputProtocolAdapter[Any]", self._adapter).extract_text(chunk)

        if not text:
            return

        self._maybe_log_stream_start()

        # Emit text delta for real-time streaming
        if self.config.emit_text_deltas:
            yield self._create_text_delta_event(text)

        # Process text through line accumulator and block state machine
        for line_number, line in self._line_accumulator.add_text(text):
            line_events = self._block_machine.process_line(line, line_number)
            self._update_stats(line_events)
            yield from line_events

    def _iter_finalize_outputs(self) -> Iterator[Event]:
        """Yield the final line's events and rejection events for incomplete blocks.

        Shared by :meth:`finalize` (consumed via ``list()``) and the tail of
        :meth:`process_stream`.
        """
        # Process any remaining accumulated text as a final line
        final_line = self._line_accumulator.finalize()
        if final_line:
            line_number, line = final_line
            line_events = self._block_machine.process_line(line, line_number)
            self._update_stats(line_events)
            yield from line_events

        # Flush remaining candidates
        flush_events = self._block_machine.flush(self._line_accumulator.line_number)
        self._update_stats(flush_events)
        yield from flush_events

    def _maybe_log_stream_start(self) -> None:
        """Log the stream-start debug event once, on the first chunk with text."""
        if self._line_accumulator.line_number == 0 and not self._line_accumulator.has_pending_text:
            self.logger.debug(
                "stream_processing_started",
                syntax=get_syntax_name(self.syntax),
                lines_buffer=self.config.lines_buffer,
                max_block_size=self.config.max_block_size,
            )

    def _create_text_delta_event(self, text: str) -> TextDeltaEvent:
        """Create a TextDeltaEvent with current block context."""
        inside_block = self._block_machine.has_active_candidates
        block_section = self._block_machine.get_current_section() if inside_block else None
        block_id = self._block_machine.get_current_block_id() if inside_block else None

        return TextDeltaEvent(
            delta=text,
            inside_block=inside_block,
            block_id=block_id,
            section=block_section,
        )

    def _update_stats(self, events: list[Event]) -> None:
        """Update stream state statistics based on events."""
        for event in events:
            if isinstance(event, BlockEndEvent):
                self._stream_state.blocks_extracted += 1
            elif isinstance(event, BlockErrorEvent):
                self._stream_state.blocks_rejected += 1

config instance-attribute

config = config or ProcessorConfig()

logger instance-attribute

logger = logger or StdlibLoggerAdapter(getLogger(__name__))

registry instance-attribute

registry = registry

syntax instance-attribute

syntax = syntax

finalize

finalize() -> list[Event]

Finalize processing and flush any incomplete blocks.

Call this method after processing all chunks to get rejection events for any blocks that were opened but never closed.

This method processes any accumulated text as a final line before flushing candidates, ensuring the last line is processed even if it doesn't end with a newline.

Returns:

Type	Description
list[Event]	List of events including processed final line and rejection events
list[Event]	for incomplete blocks

Example

processor = StreamBlockProcessor(registry) async for chunk in stream: ... events = processor.process_chunk(chunk) ... # ... handle events ...

Stream ended, process remaining text and flush incomplete blocks

final_events = processor.finalize() for event in final_events: ... if isinstance(event, BlockErrorEvent): ... print(f"Incomplete block: {event.reason}")

Source code in src/hother/streamblocks/core/processor.py

def finalize(self) -> list[Event]:
    """Finalize processing and flush any incomplete blocks.

    Call this method after processing all chunks to get rejection events
    for any blocks that were opened but never closed.

    This method processes any accumulated text as a final line before
    flushing candidates, ensuring the last line is processed even if it
    doesn't end with a newline.

    Returns:
        List of events including processed final line and rejection events
        for incomplete blocks

    Example:
        >>> processor = StreamBlockProcessor(registry)
        >>> async for chunk in stream:
        ...     events = processor.process_chunk(chunk)
        ...     # ... handle events
        ...
        >>> # Stream ended, process remaining text and flush incomplete blocks
        >>> final_events = processor.finalize()
        >>> for event in final_events:
        ...     if isinstance(event, BlockErrorEvent):
        ...         print(f"Incomplete block: {event.reason}")
    """
    return list(self._iter_finalize_outputs())

is_native_event

is_native_event(event: Any) -> bool

Check if event is a native provider event (not a StreamBlocks event).

This method provides provider-agnostic detection of native events. It checks if the event originates from the AI provider (Gemini, OpenAI, Anthropic, etc.) versus being a StreamBlocks event.

Parameters:

Name	Type	Description	Default
event	Any	Event to check	required

Returns:

Type	Description
bool	True if event is from the native provider, False if it's a StreamBlocks
bool	event or if detection is not possible

Example

processor = StreamBlockProcessor(registry) async for event in processor.process_stream(gemini_stream): ... if processor.is_native_event(event): ... # Handle Gemini event (provider-agnostic!) ... usage = getattr(event, 'usage_metadata', None) ... elif isinstance(event, BlockEndEvent): ... # Handle StreamBlocks event ... print(f"Block: {event.block_id}")

Source code in src/hother/streamblocks/core/processor.py

def is_native_event(self, event: Any) -> bool:
    """Check if event is a native provider event (not a StreamBlocks event).

    This method provides provider-agnostic detection of native events.
    It checks if the event originates from the AI provider (Gemini, OpenAI,
    Anthropic, etc.) versus being a StreamBlocks event.

    Args:
        event: Event to check

    Returns:
        True if event is from the native provider, False if it's a StreamBlocks
        event or if detection is not possible

    Example:
        >>> processor = StreamBlockProcessor(registry)
        >>> async for event in processor.process_stream(gemini_stream):
        ...     if processor.is_native_event(event):
        ...         # Handle Gemini event (provider-agnostic!)
        ...         usage = getattr(event, 'usage_metadata', None)
        ...     elif isinstance(event, BlockEndEvent):
        ...         # Handle StreamBlocks event
        ...         print(f"Block: {event.block_id}")
    """
    # Check if it's a known StreamBlocks event
    if isinstance(
        event,
        (
            StreamStartedEvent,
            StreamFinishedEvent,
            TextContentEvent,
            TextDeltaEvent,
            BlockStartEvent,
            BlockHeaderDeltaEvent,
            BlockMetadataDeltaEvent,
            BlockContentDeltaEvent,
            BlockMetadataEndEvent,
            BlockContentEndEvent,
            BlockEndEvent,
            BlockErrorEvent,
        ),
    ):
        return False

    # Check if we have an adapter with a module prefix
    if self._adapter is None:
        return False

    # Use Protocol-based check for native module prefix
    if not isinstance(self._adapter, HasNativeModulePrefix):
        return False

    # Check if event's module matches the adapter's prefix
    return type(event).__module__.startswith(self._adapter.native_module_prefix)

process_chunk

process_chunk(
    chunk: TChunk,
    adapter: InputProtocolAdapter[TChunk] | None = None,
) -> list[TChunk | Event]

Process a single chunk and return resulting events.

This method is stateful - it maintains internal state between calls. Call finalize() after processing all chunks to flush incomplete blocks.

Parameters:

Name	Type	Description	Default
chunk	TChunk	Single chunk to process	required
adapter	InputProtocolAdapter[TChunk] | None	Optional adapter for extracting text. If not provided and auto_detect_adapter=True, will auto-detect on first chunk.	None

Returns:

Type	Description
list[TChunk | Event]	List of events generated from this chunk. May be empty if chunk only
list[TChunk | Event]	accumulates text without completing any lines.

Raises:

Type	Description
RuntimeError	If adapter is not set after first chunk processing (internal state error, should not occur in normal usage).

Example

processor = StreamBlockProcessor(registry) response = await client.generate_content_stream(...) async for chunk in response: ... events = processor.process_chunk(chunk) ... for event in events: ... if isinstance(event, BlockEndEvent): ... print(f"Block: {event.block_id}") ...

Finalize at stream end

final_events = processor.finalize() for event in final_events: ... if isinstance(event, BlockErrorEvent): ... print(f"Incomplete block: {event.reason}")

Source code in src/hother/streamblocks/core/processor.py

def process_chunk(
    self,
    chunk: TChunk,
    adapter: InputProtocolAdapter[TChunk] | None = None,
) -> list[TChunk | Event]:
    """Process a single chunk and return resulting events.

    This method is stateful - it maintains internal state between calls.
    Call finalize() after processing all chunks to flush incomplete blocks.

    Args:
        chunk: Single chunk to process
        adapter: Optional adapter for extracting text. If not provided and
                auto_detect_adapter=True, will auto-detect on first chunk.

    Returns:
        List of events generated from this chunk. May be empty if chunk only
        accumulates text without completing any lines.

    Raises:
        RuntimeError: If adapter is not set after first chunk processing
            (internal state error, should not occur in normal usage).

    Example:
        >>> processor = StreamBlockProcessor(registry)
        >>> response = await client.generate_content_stream(...)
        >>> async for chunk in response:
        ...     events = processor.process_chunk(chunk)
        ...     for event in events:
        ...         if isinstance(event, BlockEndEvent):
        ...             print(f"Block: {event.block_id}")
        ...
        >>> # Finalize at stream end
        >>> final_events = processor.finalize()
        >>> for event in final_events:
        ...     if isinstance(event, BlockErrorEvent):
        ...         print(f"Incomplete block: {event.reason}")
    """
    return list(self._iter_chunk_outputs(chunk, adapter, context="process_chunk"))

process_stream async

process_stream(
    stream: AsyncIterator[TChunk],
    adapter: InputProtocolAdapter[TChunk] | None = None,
) -> AsyncGenerator[TChunk | Event]

Process stream and yield mixed events.

This method processes chunks from any stream format, extracting text via an adapter and emitting both original chunks (if enabled) and StreamBlocks events.

Parameters:

Name	Type	Description	Default
stream	AsyncIterator[TChunk]	Async iterator yielding chunks (text or objects)	required
adapter	InputProtocolAdapter[TChunk] | None	Optional adapter for extracting text from chunks. If None and auto_detect_adapter=True, will auto-detect from first chunk.	None

Yields:

Type	Description
AsyncGenerator[TChunk | Event]	Mixed stream of:
AsyncGenerator[TChunk | Event]	Original chunks (if emit_original_events=True)
AsyncGenerator[TChunk | Event]	TextDeltaEvent (if emit_text_deltas=True)
AsyncGenerator[TChunk | Event]	TextContentEvent, BlockStartEvent, BlockEndEvent, BlockErrorEvent, and section delta events

Raises:

Type	Description
RuntimeError	If adapter is not set after first chunk processing (internal state error, should not occur in normal usage).

Example

Plain text

async for event in processor.process_stream(text_stream): ... if isinstance(event, BlockEndEvent): ... print(f"Extracted: {event.block_id}")

With Gemini adapter (auto-detected)

async for event in processor.process_stream(gemini_stream): ... if hasattr(event, 'usage_metadata'): ... print(f"Tokens: {event.usage_metadata}") ... elif isinstance(event, BlockEndEvent): ... print(f"Extracted: {event.block_id}")

Source code in src/hother/streamblocks/core/processor.py

async def process_stream(
    self,
    stream: AsyncIterator[TChunk],
    adapter: InputProtocolAdapter[TChunk] | None = None,
) -> AsyncGenerator[TChunk | Event]:
    """Process stream and yield mixed events.

    This method processes chunks from any stream format, extracting text
    via an adapter and emitting both original chunks (if enabled) and
    StreamBlocks events.

    Args:
        stream: Async iterator yielding chunks (text or objects)
        adapter: Optional adapter for extracting text from chunks.
                If None and auto_detect_adapter=True, will auto-detect from first chunk.

    Yields:
        Mixed stream of:
        - Original chunks (if emit_original_events=True)
        - TextDeltaEvent (if emit_text_deltas=True)
        - TextContentEvent, BlockStartEvent, BlockEndEvent, BlockErrorEvent, and section delta events

    Raises:
        RuntimeError: If adapter is not set after first chunk processing
            (internal state error, should not occur in normal usage).

    Example:
        >>> # Plain text
        >>> async for event in processor.process_stream(text_stream):
        ...     if isinstance(event, BlockEndEvent):
        ...         print(f"Extracted: {event.block_id}")
        >>>
        >>> # With Gemini adapter (auto-detected)
        >>> async for event in processor.process_stream(gemini_stream):
        ...     if hasattr(event, 'usage_metadata'):
        ...         print(f"Tokens: {event.usage_metadata}")
        ...     elif isinstance(event, BlockEndEvent):
        ...         print(f"Extracted: {event.block_id}")
    """
    # Set adapter if provided
    if adapter:
        self._adapter = adapter
        self._first_chunk_processed = True

    async for chunk in stream:
        for output in self._iter_chunk_outputs(chunk, None, context="process_stream"):
            yield output

    # Process remaining text and flush incomplete blocks at stream end
    for event in self._iter_finalize_outputs():
        yield event

StreamState dataclass

Tracks state during stream processing.

Source code in src/hother/streamblocks/core/processor.py

@dataclass
class StreamState:
    """Tracks state during stream processing."""

    stream_id: str = field(default_factory=lambda: str(uuid4()))
    start_time: int = field(default_factory=lambda: int(time.time() * 1000))
    blocks_extracted: int = 0
    blocks_rejected: int = 0
    total_events: int = 0

    def duration_ms(self) -> int:
        """Get duration in milliseconds since stream started."""
        return int(time.time() * 1000) - self.start_time

blocks_extracted class-attribute instance-attribute

blocks_extracted: int = 0

blocks_rejected class-attribute instance-attribute

blocks_rejected: int = 0

start_time class-attribute instance-attribute

start_time: int = field(
    default_factory=lambda: int(time() * 1000)
)

stream_id class-attribute instance-attribute

stream_id: str = field(default_factory=lambda: str(uuid4()))

total_events class-attribute instance-attribute

total_events: int = 0

duration_ms

duration_ms() -> int

Get duration in milliseconds since stream started.

Source code in src/hother/streamblocks/core/processor.py

def duration_ms(self) -> int:
    """Get duration in milliseconds since stream started."""
    return int(time.time() * 1000) - self.start_time

hother.streamblocks.core.protocol_processor

Protocol stream processor for bidirectional adapter support.

TInput module-attribute

TInput = TypeVar('TInput')

TOutput module-attribute

TOutput = TypeVar('TOutput')

ProtocolStreamProcessor

StreamBlocks processor with bidirectional protocol support.

This processor enables processing of any input stream format and transformation to any output format using the adapter pattern.

The processor supports three usage modes:

1. Auto-detect input: Pass input_adapter=None to auto-detect from first event
2. Explicit adapters: Specify both input and output adapters
3. Default output: Pass output_adapter=None to emit native StreamBlocks events

Example

from hother.streamblocks import ProtocolStreamProcessor, Registry from hother.streamblocks.adapters.input import IdentityInputAdapter from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter

Auto-detect input, native output

processor = ProtocolStreamProcessor(registry)

Explicit adapters

processor = ProtocolStreamProcessor( ... registry, ... input_adapter=IdentityInputAdapter(), ... output_adapter=StreamBlocksOutputAdapter(), ... )

Process stream

async for event in processor.process_stream(input_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/core/protocol_processor.py

class ProtocolStreamProcessor[TInput, TOutput]:
    """StreamBlocks processor with bidirectional protocol support.

    This processor enables processing of any input stream format and transformation
    to any output format using the adapter pattern.

    The processor supports three usage modes:

    1. **Auto-detect input**: Pass `input_adapter=None` to auto-detect from first event
    2. **Explicit adapters**: Specify both input and output adapters
    3. **Default output**: Pass `output_adapter=None` to emit native StreamBlocks events

    Example:
        >>> from hother.streamblocks import ProtocolStreamProcessor, Registry
        >>> from hother.streamblocks.adapters.input import IdentityInputAdapter
        >>> from hother.streamblocks.adapters.output import StreamBlocksOutputAdapter
        >>>
        >>> # Auto-detect input, native output
        >>> processor = ProtocolStreamProcessor(registry)
        >>>
        >>> # Explicit adapters
        >>> processor = ProtocolStreamProcessor(
        ...     registry,
        ...     input_adapter=IdentityInputAdapter(),
        ...     output_adapter=StreamBlocksOutputAdapter(),
        ... )
        >>>
        >>> # Process stream
        >>> async for event in processor.process_stream(input_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """

    def __init__(
        self,
        registry: Registry,
        input_adapter: InputProtocolAdapter[TInput] | None = None,
        output_adapter: OutputProtocolAdapter[TOutput] | None = None,
        config: ProcessorConfig | None = None,
        *,
        logger: Logger | None = None,
    ) -> None:
        """Initialize the protocol stream processor.

        Args:
            registry: Registry with syntax and block definitions
            input_adapter: Input adapter for extracting text from events.
                          If None, will auto-detect from first event.
            output_adapter: Output adapter for transforming StreamBlocks events.
                           If None, will emit native StreamBlocks events.
            config: Configuration object for processor settings
            logger: Optional logger
        """
        self.registry = registry
        self._input_adapter = input_adapter
        # StreamBlocksOutputAdapter emits native BaseEvent output and cannot be
        # statically proven to satisfy OutputProtocolAdapter[TOutput] for an arbitrary
        # TOutput, so cast the default rather than suppress.
        self._output_adapter: OutputProtocolAdapter[TOutput] = cast(
            "OutputProtocolAdapter[TOutput]",
            output_adapter if output_adapter is not None else StreamBlocksOutputAdapter(),
        )
        self._auto_detected = False

        # Store config, using modified defaults for protocol processing
        self._config = config or ProcessorConfig(
            emit_original_events=False,  # We handle original events via passthrough
            auto_detect_adapter=False,  # We handle detection ourselves
        )

        # Create internal core processor
        self._core_processor = StreamBlockProcessor(
            registry,
            config=self._config,
            logger=logger,
        )

    @property
    def was_auto_detected(self) -> bool:
        """Whether the input adapter was auto-detected.

        Returns:
            True if adapter was detected from first event, False if explicitly provided
        """
        return self._auto_detected

    @property
    def input_adapter(self) -> InputProtocolAdapter[TInput] | None:
        """The active input adapter (may be auto-detected).

        Returns:
            The input adapter, or None if not yet detected
        """
        return self._input_adapter

    @property
    def output_adapter(self) -> OutputProtocolAdapter[TOutput]:
        """The active output adapter.

        Returns:
            The output adapter
        """
        return self._output_adapter

    async def process_stream(
        self,
        input_stream: AsyncIterator[TInput],
    ) -> AsyncIterator[TOutput]:
        """Process input stream through StreamBlocks, emit output protocol events.

        This method:
        1. Auto-detects input adapter on first event if not specified
        2. Categorizes each event (TEXT_CONTENT, PASSTHROUGH, SKIP)
        3. For TEXT_CONTENT: extracts text, processes through StreamBlocks, transforms output
        4. For PASSTHROUGH: passes event through output adapter's passthrough method
        5. For SKIP: doesn't emit anything

        Args:
            input_stream: Async iterator yielding input events

        Yields:
            Output protocol events

        Example:
            >>> async for event in processor.process_stream(openai_stream):
            ...     if isinstance(event, BlockExtractedEvent):
            ...         print(f"Block: {event.block.metadata.id}")
        """
        async for input_event in input_stream:
            async for output_event in self._process_input_event(input_event):
                yield output_event

        # Finalize stream processing
        async for output_event in self._finalize_stream():
            yield output_event

    async def _process_input_event(
        self,
        input_event: TInput,
    ) -> AsyncIterator[TOutput]:
        """Process a single input event and yield output events."""
        # Auto-detect input adapter on first event if not specified
        if self._input_adapter is None:
            self._input_adapter = detect_input_adapter(input_event)
            self._auto_detected = True

        category = self._input_adapter.categorize(input_event)

        if category == EventCategory.TEXT_CONTENT:
            async for event in self._process_text_content(input_event):
                yield event
        elif category == EventCategory.PASSTHROUGH:
            output = self._output_adapter.passthrough(input_event)
            if output is not None:
                yield output
        # SKIP category: Don't emit anything

    async def _process_text_content(
        self,
        input_event: TInput,
    ) -> AsyncIterator[TOutput]:
        """Process TEXT_CONTENT event and yield output events."""
        # _input_adapter is guaranteed to be set by _process_input_event
        if self._input_adapter is None:  # pragma: no cover
            raise AdapterNotConfiguredError(context="protocol_processor")
        text = self._input_adapter.extract_text(input_event)
        if text:
            for sb_event in self._core_processor.process_chunk(text):
                async for event in self._transform_sb_event(sb_event):
                    yield event

    async def _transform_sb_event(
        self,
        sb_event: Event | str,
    ) -> AsyncIterator[TOutput]:
        """Transform StreamBlocks event to output protocol events."""
        if isinstance(sb_event, str):  # pragma: no cover - emit_original_events is False here
            # Original passthrough chunks are disabled for protocol processing,
            # so only StreamBlocks events reach this transform.
            return
        output = self._output_adapter.to_protocol_event(sb_event)
        if output is not None:
            async for event in self._ensure_async_iterable(output):
                yield event

    async def _finalize_stream(self) -> AsyncIterator[TOutput]:
        """Finalize stream and yield remaining output events."""
        for sb_event in self._core_processor.finalize():
            async for event in self._transform_sb_event(sb_event):
                yield event

    def process_chunk(
        self,
        input_event: TInput,
    ) -> list[TOutput]:
        """Process a single input event synchronously.

        This method is stateful - it maintains internal state between calls.
        Call finalize() after processing all events to flush incomplete blocks.

        Args:
            input_event: Single input event to process

        Returns:
            List of output events generated from this input event
        """
        events: list[TOutput] = []

        # Auto-detect input adapter on first event if not specified
        if self._input_adapter is None:
            self._input_adapter = detect_input_adapter(input_event)
            self._auto_detected = True

        # Categorize the event
        category = self._input_adapter.categorize(input_event)

        if category == EventCategory.TEXT_CONTENT:
            # Extract text and process through StreamBlocks
            text = self._input_adapter.extract_text(input_event)
            if text:
                # Process text chunk and get StreamBlocks events
                for sb_event in self._core_processor.process_chunk(text):
                    if isinstance(sb_event, str):  # pragma: no cover - emit_original_events is False here
                        continue
                    output = self._output_adapter.to_protocol_event(sb_event)
                    if output is not None:
                        events.extend(self._ensure_list(output))

        elif category == EventCategory.PASSTHROUGH:
            # Pass through to output adapter
            output = self._output_adapter.passthrough(input_event)
            if output is not None:
                events.append(output)

        # SKIP category: Don't emit anything

        return events

    def finalize(self) -> list[TOutput]:
        """Finalize processing and flush any incomplete blocks.

        Call this method after processing all events to get rejection events
        for any blocks that were opened but never closed.

        Returns:
            List of output events including rejection events for incomplete blocks
        """
        events: list[TOutput] = []

        for sb_event in self._core_processor.finalize():
            output = self._output_adapter.to_protocol_event(sb_event)
            if output is not None:
                events.extend(self._ensure_list(output))

        return events

    def reset(self) -> None:
        """Reset the processor state for reuse.

        This clears all internal state including:
        - Auto-detected adapter (will re-detect on next stream)
        - Core processor buffers and candidates
        """
        self._input_adapter = None
        self._auto_detected = False
        # Recreate core processor to reset its state
        self._core_processor = StreamBlockProcessor(
            self.registry,
            config=self._config,
            logger=self._core_processor.logger,
        )

    async def _ensure_async_iterable(
        self,
        output: TOutput | list[TOutput],
    ) -> AsyncIterator[TOutput]:
        """Ensure output is iterable, yielding single or multiple events.

        Args:
            output: Single event or list of events

        Yields:
            Individual events
        """
        if isinstance(output, list):
            for item in cast("list[TOutput]", output):
                yield item
        else:
            yield output

    def _ensure_list(self, output: TOutput | list[TOutput]) -> list[TOutput]:
        """Ensure output is a list.

        Args:
            output: Single event or list of events

        Returns:
            List of events
        """
        if isinstance(output, list):
            # See _ensure_async_iterable: narrow the unbounded-TypeVar list element type.
            return cast("list[TOutput]", output)
        return [output]

input_adapter property

input_adapter: InputProtocolAdapter[TInput] | None

The active input adapter (may be auto-detected).

Returns:

Type	Description
InputProtocolAdapter[TInput] | None	The input adapter, or None if not yet detected

output_adapter property

output_adapter: OutputProtocolAdapter[TOutput]

The active output adapter.

Returns:

Type	Description
OutputProtocolAdapter[TOutput]	The output adapter

registry instance-attribute

registry = registry

was_auto_detected property

was_auto_detected: bool

Whether the input adapter was auto-detected.

Returns:

Type	Description
bool	True if adapter was detected from first event, False if explicitly provided

finalize

finalize() -> list[TOutput]

Finalize processing and flush any incomplete blocks.

Call this method after processing all events to get rejection events for any blocks that were opened but never closed.

Returns:

Type	Description
list[TOutput]	List of output events including rejection events for incomplete blocks

Source code in src/hother/streamblocks/core/protocol_processor.py

def finalize(self) -> list[TOutput]:
    """Finalize processing and flush any incomplete blocks.

    Call this method after processing all events to get rejection events
    for any blocks that were opened but never closed.

    Returns:
        List of output events including rejection events for incomplete blocks
    """
    events: list[TOutput] = []

    for sb_event in self._core_processor.finalize():
        output = self._output_adapter.to_protocol_event(sb_event)
        if output is not None:
            events.extend(self._ensure_list(output))

    return events

process_chunk

process_chunk(input_event: TInput) -> list[TOutput]

Process a single input event synchronously.

This method is stateful - it maintains internal state between calls. Call finalize() after processing all events to flush incomplete blocks.

Parameters:

Name	Type	Description	Default
input_event	TInput	Single input event to process	required

Returns:

Type	Description
list[TOutput]	List of output events generated from this input event

Source code in src/hother/streamblocks/core/protocol_processor.py

def process_chunk(
    self,
    input_event: TInput,
) -> list[TOutput]:
    """Process a single input event synchronously.

    This method is stateful - it maintains internal state between calls.
    Call finalize() after processing all events to flush incomplete blocks.

    Args:
        input_event: Single input event to process

    Returns:
        List of output events generated from this input event
    """
    events: list[TOutput] = []

    # Auto-detect input adapter on first event if not specified
    if self._input_adapter is None:
        self._input_adapter = detect_input_adapter(input_event)
        self._auto_detected = True

    # Categorize the event
    category = self._input_adapter.categorize(input_event)

    if category == EventCategory.TEXT_CONTENT:
        # Extract text and process through StreamBlocks
        text = self._input_adapter.extract_text(input_event)
        if text:
            # Process text chunk and get StreamBlocks events
            for sb_event in self._core_processor.process_chunk(text):
                if isinstance(sb_event, str):  # pragma: no cover - emit_original_events is False here
                    continue
                output = self._output_adapter.to_protocol_event(sb_event)
                if output is not None:
                    events.extend(self._ensure_list(output))

    elif category == EventCategory.PASSTHROUGH:
        # Pass through to output adapter
        output = self._output_adapter.passthrough(input_event)
        if output is not None:
            events.append(output)

    # SKIP category: Don't emit anything

    return events

process_stream async

process_stream(
    input_stream: AsyncIterator[TInput],
) -> AsyncIterator[TOutput]

Process input stream through StreamBlocks, emit output protocol events.

This method: 1. Auto-detects input adapter on first event if not specified 2. Categorizes each event (TEXT_CONTENT, PASSTHROUGH, SKIP) 3. For TEXT_CONTENT: extracts text, processes through StreamBlocks, transforms output 4. For PASSTHROUGH: passes event through output adapter's passthrough method 5. For SKIP: doesn't emit anything

Parameters:

Name	Type	Description	Default
input_stream	AsyncIterator[TInput]	Async iterator yielding input events	required

Yields:

Type	Description
AsyncIterator[TOutput]	Output protocol events

Example

async for event in processor.process_stream(openai_stream): ... if isinstance(event, BlockExtractedEvent): ... print(f"Block: {event.block.metadata.id}")

Source code in src/hother/streamblocks/core/protocol_processor.py

async def process_stream(
    self,
    input_stream: AsyncIterator[TInput],
) -> AsyncIterator[TOutput]:
    """Process input stream through StreamBlocks, emit output protocol events.

    This method:
    1. Auto-detects input adapter on first event if not specified
    2. Categorizes each event (TEXT_CONTENT, PASSTHROUGH, SKIP)
    3. For TEXT_CONTENT: extracts text, processes through StreamBlocks, transforms output
    4. For PASSTHROUGH: passes event through output adapter's passthrough method
    5. For SKIP: doesn't emit anything

    Args:
        input_stream: Async iterator yielding input events

    Yields:
        Output protocol events

    Example:
        >>> async for event in processor.process_stream(openai_stream):
        ...     if isinstance(event, BlockExtractedEvent):
        ...         print(f"Block: {event.block.metadata.id}")
    """
    async for input_event in input_stream:
        async for output_event in self._process_input_event(input_event):
            yield output_event

    # Finalize stream processing
    async for output_event in self._finalize_stream():
        yield output_event

reset

reset() -> None

Reset the processor state for reuse.

This clears all internal state including: - Auto-detected adapter (will re-detect on next stream) - Core processor buffers and candidates

Source code in src/hother/streamblocks/core/protocol_processor.py

def reset(self) -> None:
    """Reset the processor state for reuse.

    This clears all internal state including:
    - Auto-detected adapter (will re-detect on next stream)
    - Core processor buffers and candidates
    """
    self._input_adapter = None
    self._auto_detected = False
    # Recreate core processor to reset its state
    self._core_processor = StreamBlockProcessor(
        self.registry,
        config=self._config,
        logger=self._core_processor.logger,
    )