Registry & Models

The registry that maps block types to block classes, and the block data models. See Blocks & Registry for the conceptual guide.

hother.streamblocks.core.registry

Type-specific registry for StreamBlocks.

BlockType

BlockType = str

ContentValidatorFunc

ContentValidatorFunc = Callable[
    [str, dict[str, Any] | None], ValidationResult
]

MetadataValidatorFunc

MetadataValidatorFunc = Callable[
    [str, dict[str, Any] | None], ValidationResult
]

SyntaxName

SyntaxName = str

ValidatorFunc

ValidatorFunc = Callable[[ExtractedBlock[Any, Any]], bool]

MetadataValidationFailureMode

Bases: StrEnum

Behavior when metadata validation fails.

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

class MetadataValidationFailureMode(StrEnum):
    """Behavior when metadata validation fails."""

    ABORT_BLOCK = "abort_block"  # Emit BlockErrorEvent immediately
    CONTINUE = "continue"  # Continue with warning, process content
    SKIP_CONTENT = "skip_content"  # Skip content, emit partial block

ABORT_BLOCK class-attribute instance-attribute

ABORT_BLOCK = 'abort_block'

CONTINUE class-attribute instance-attribute

CONTINUE = 'continue'

SKIP_CONTENT class-attribute instance-attribute

SKIP_CONTENT = 'skip_content'

Registry

Type-specific registry for a single syntax type.

This registry holds exactly one syntax instance and maps block types to block classes.

Example

syntax = DelimiterPreambleSyntax(name="my_syntax") registry = Registry(syntax) registry.register("files_operations", FileOperations, validators=[my_validator]) registry.register("patch", Patch)

Or with bulk registration:

registry = Registry( ... syntax=syntax, ... blocks={ ... "files_operations": FileOperations, ... "patch": Patch, ... } ... ) registry.add_validator("files_operations", my_validator)

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

class Registry:
    """Type-specific registry for a single syntax type.

    This registry holds exactly one syntax instance and maps block types to block classes.

    Example:
        >>> syntax = DelimiterPreambleSyntax(name="my_syntax")
        >>> registry = Registry(syntax)
        >>> registry.register("files_operations", FileOperations, validators=[my_validator])
        >>> registry.register("patch", Patch)

        Or with bulk registration:
        >>> registry = Registry(
        ...     syntax=syntax,
        ...     blocks={
        ...         "files_operations": FileOperations,
        ...         "patch": Patch,
        ...     }
        ... )
        >>> registry.add_validator("files_operations", my_validator)
    """

    def __init__(
        self,
        *,
        syntax: Syntax | BaseSyntax = Syntax.DELIMITER_PREAMBLE,
        logger: Logger | None = None,
        blocks: dict[str, type[Block[Any, Any]]] | None = None,
        metadata_failure_mode: MetadataValidationFailureMode = MetadataValidationFailureMode.ABORT_BLOCK,
    ) -> None:
        """Initialize registry with a single syntax instance.

        Args:
            syntax: The syntax instance for this registry.
                   Defaults to Syntax.DELIMITER_PREAMBLE.
            logger: Optional logger (any object with debug/info/warning/error/exception methods).
                   Defaults to stdlib logging.getLogger(__name__)
            blocks: Optional dict of block_type -> block_class for bulk registration
            metadata_failure_mode: Behavior when metadata validation fails
        """
        self._syntax = get_syntax_instance(syntax=syntax)
        self._block_classes: dict[BlockType, type[Block[Any, Any]]] = {}
        self._validators: dict[BlockType, list[ValidatorFunc]] = {}
        self._metadata_validators: dict[BlockType, list[MetadataValidatorFunc]] = {}
        self._content_validators: dict[BlockType, list[ContentValidatorFunc]] = {}
        self._metadata_failure_mode = metadata_failure_mode
        self._template_manager: TemplateManager | None = None
        self.logger = logger or StdlibLoggerAdapter(logging.getLogger(__name__))

        # Bulk register blocks if provided
        if blocks:
            for block_type, block_class in blocks.items():
                self.register(block_type, block_class)

    @property
    def syntax(self) -> BaseSyntax:
        """Get the syntax instance."""
        return self._syntax

    @property
    def registered_blocks(self) -> Mapping[str, type[Block[Any, Any]]]:
        """Read-only view of registered block types mapped to block classes."""
        return MappingProxyType(self._block_classes)

    def to_prompt(self, *, include_examples: bool = True, template_version: str = "default") -> str:
        """Generate an LLM system prompt documenting all registered blocks.

        The prompt describes this registry's syntax format and, for each
        registered block type, its description, metadata fields, content
        format, and (optionally) serialized examples.

        Args:
            include_examples: Whether to render each block's examples.
            template_version: Template version for A/B testing.

        Returns:
            The rendered system prompt.
        """
        blocks = [
            build_block_context(block_class, self._syntax, include_examples=include_examples)
            for block_class in self._block_classes.values()
        ]
        context: dict[str, Any] = {
            "syntax_name": type(self._syntax).__name__,
            "syntax_format": self._syntax.describe_format(),
            "blocks": blocks,
        }
        return self._get_template_manager().render(context, template_version, mode="registry")

    def register_template(self, version: str, template: str | Path, mode: str = "both") -> None:
        """Register a custom prompt template for this registry.

        Args:
            version: Version identifier used by ``to_prompt(template_version=...)``.
            template: Template string, or a Path to a template file.
            mode: "registry", "single", or "both".
        """
        self._get_template_manager().register_template(version, template, mode)

    def serialize_block(self, block: Block[Any, Any]) -> str:
        """Serialize a block instance using this registry's syntax."""
        return self._syntax.serialize_block(block)

    def _get_template_manager(self) -> TemplateManager:
        """Lazily create the per-registry template manager."""
        if self._template_manager is None:
            self._template_manager = TemplateManager()
        return self._template_manager

    def register(
        self,
        name: str,
        block_class: type[Block[Any, Any]],
        validators: list[ValidatorFunc] | None = None,
    ) -> None:
        """Register a block class for a block type.

        Args:
            name: Block type name (e.g., "files_operations", "patch")
            block_class: Block class inheriting from Block[M, C]
            validators: Optional list of validator functions for this block type
        """
        self._block_classes[name] = block_class

        self.logger.debug(
            "block_type_registered",
            block_type=name,
            block_class=block_class.__name__,
            has_validators=validators is not None and len(validators) > 0,
        )

        # Add validators if provided
        if validators:
            for validator in validators:
                self.add_validator(name, validator)

    def get_block_class(self, block_type: str) -> type[Block[Any, Any]] | None:
        """Get the block class for a given block type.

        Args:
            block_type: The block type to look up

        Returns:
            The registered block class, or None if not found
        """
        return self._block_classes.get(block_type)

    def add_validator(
        self,
        block_type: BlockType,
        validator: ValidatorFunc,
    ) -> None:
        """Add a validator for a block type.

        Args:
            block_type: Type of block to validate
            validator: Function that validates a block
        """
        if block_type not in self._validators:
            self._validators[block_type] = []
        self._validators[block_type].append(validator)

        self.logger.debug(
            "validator_added",
            block_type=block_type,
            validator_name=validator.__name__,
            total_validators=len(self._validators[block_type]),
        )

    def validate_block(
        self,
        block: ExtractedBlock[Any, Any],
    ) -> bool:
        """Run all validators for a block.

        Args:
            block: Extracted block to validate

        Returns:
            True if all validators pass
        """
        # block.metadata is BaseMetadata which always has block_type
        block_type = block.metadata.block_type
        if not block_type:
            return True

        validators = self._validators.get(block_type, [])
        return all(v(block) for v in validators)

    @property
    def metadata_failure_mode(self) -> MetadataValidationFailureMode:
        """Get the metadata validation failure mode."""
        return self._metadata_failure_mode

    def add_metadata_validator(
        self,
        block_type: BlockType,
        validator: MetadataValidatorFunc,
    ) -> None:
        """Add an early metadata validator for a block type.

        Metadata validators are called when the metadata section completes,
        before content accumulation begins. They receive the raw metadata
        string and the parsed metadata dict.

        Args:
            block_type: Type of block to validate
            validator: Function that validates metadata and returns ValidationResult
        """
        if block_type not in self._metadata_validators:
            self._metadata_validators[block_type] = []
        self._metadata_validators[block_type].append(validator)

        self.logger.debug(
            "metadata_validator_added",
            block_type=block_type,
            validator_name=validator.__name__,
            total_validators=len(self._metadata_validators[block_type]),
        )

    def add_content_validator(
        self,
        block_type: BlockType,
        validator: ContentValidatorFunc,
    ) -> None:
        """Add an early content validator for a block type.

        Content validators are called when the content section completes,
        before the final BlockEndEvent. They receive the raw content
        string and the parsed content dict.

        Args:
            block_type: Type of block to validate
            validator: Function that validates content and returns ValidationResult
        """
        if block_type not in self._content_validators:
            self._content_validators[block_type] = []
        self._content_validators[block_type].append(validator)

        self.logger.debug(
            "content_validator_added",
            block_type=block_type,
            validator_name=validator.__name__,
            total_validators=len(self._content_validators[block_type]),
        )

    def validate_metadata(
        self,
        block_type: str,
        raw_metadata: str,
        parsed_metadata: dict[str, Any] | None,
    ) -> ValidationResult:
        """Run all metadata validators for a block type.

        Args:
            block_type: Type of block being validated
            raw_metadata: Raw metadata string
            parsed_metadata: Parsed metadata dict (if available)

        Returns:
            ValidationResult with combined results from all validators
        """
        validators = self._metadata_validators.get(block_type, [])
        if not validators:
            return ValidationResult.success()

        for validator in validators:
            result = validator(raw_metadata, parsed_metadata)
            if not result.passed:
                self.logger.debug(
                    "metadata_validation_failed",
                    block_type=block_type,
                    validator_name=validator.__name__,
                    error=result.error,
                )
                return result

        return ValidationResult.success()

    def validate_content(
        self,
        block_type: str,
        raw_content: str,
        parsed_content: dict[str, Any] | None,
    ) -> ValidationResult:
        """Run all content validators for a block type.

        Args:
            block_type: Type of block being validated
            raw_content: Raw content string
            parsed_content: Parsed content dict (if available)

        Returns:
            ValidationResult with combined results from all validators
        """
        validators = self._content_validators.get(block_type, [])
        if not validators:
            return ValidationResult.success()

        for validator in validators:
            result = validator(raw_content, parsed_content)
            if not result.passed:
                self.logger.debug(
                    "content_validation_failed",
                    block_type=block_type,
                    validator_name=validator.__name__,
                    error=result.error,
                )
                return result

        return ValidationResult.success()

logger instance-attribute

logger = logger or StdlibLoggerAdapter(getLogger(__name__))

metadata_failure_mode property

metadata_failure_mode: MetadataValidationFailureMode

Get the metadata validation failure mode.

registered_blocks property

registered_blocks: Mapping[str, type[Block[Any, Any]]]

Read-only view of registered block types mapped to block classes.

syntax property

syntax: BaseSyntax

Get the syntax instance.

add_content_validator

add_content_validator(
    block_type: BlockType, validator: ContentValidatorFunc
) -> None

Add an early content validator for a block type.

Content validators are called when the content section completes, before the final BlockEndEvent. They receive the raw content string and the parsed content dict.

Parameters:

Name	Type	Description	Default
block_type	BlockType	Type of block to validate	required
validator	ContentValidatorFunc	Function that validates content and returns ValidationResult	required

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

def add_content_validator(
    self,
    block_type: BlockType,
    validator: ContentValidatorFunc,
) -> None:
    """Add an early content validator for a block type.

    Content validators are called when the content section completes,
    before the final BlockEndEvent. They receive the raw content
    string and the parsed content dict.

    Args:
        block_type: Type of block to validate
        validator: Function that validates content and returns ValidationResult
    """
    if block_type not in self._content_validators:
        self._content_validators[block_type] = []
    self._content_validators[block_type].append(validator)

    self.logger.debug(
        "content_validator_added",
        block_type=block_type,
        validator_name=validator.__name__,
        total_validators=len(self._content_validators[block_type]),
    )

add_metadata_validator

add_metadata_validator(
    block_type: BlockType, validator: MetadataValidatorFunc
) -> None

Add an early metadata validator for a block type.

Metadata validators are called when the metadata section completes, before content accumulation begins. They receive the raw metadata string and the parsed metadata dict.

Parameters:

Name	Type	Description	Default
block_type	BlockType	Type of block to validate	required
validator	MetadataValidatorFunc	Function that validates metadata and returns ValidationResult	required

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

def add_metadata_validator(
    self,
    block_type: BlockType,
    validator: MetadataValidatorFunc,
) -> None:
    """Add an early metadata validator for a block type.

    Metadata validators are called when the metadata section completes,
    before content accumulation begins. They receive the raw metadata
    string and the parsed metadata dict.

    Args:
        block_type: Type of block to validate
        validator: Function that validates metadata and returns ValidationResult
    """
    if block_type not in self._metadata_validators:
        self._metadata_validators[block_type] = []
    self._metadata_validators[block_type].append(validator)

    self.logger.debug(
        "metadata_validator_added",
        block_type=block_type,
        validator_name=validator.__name__,
        total_validators=len(self._metadata_validators[block_type]),
    )

add_validator

add_validator(
    block_type: BlockType, validator: ValidatorFunc
) -> None

Add a validator for a block type.

Parameters:

Name	Type	Description	Default
block_type	BlockType	Type of block to validate	required
validator	ValidatorFunc	Function that validates a block	required

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

def add_validator(
    self,
    block_type: BlockType,
    validator: ValidatorFunc,
) -> None:
    """Add a validator for a block type.

    Args:
        block_type: Type of block to validate
        validator: Function that validates a block
    """
    if block_type not in self._validators:
        self._validators[block_type] = []
    self._validators[block_type].append(validator)

    self.logger.debug(
        "validator_added",
        block_type=block_type,
        validator_name=validator.__name__,
        total_validators=len(self._validators[block_type]),
    )

get_block_class

get_block_class(
    block_type: str,
) -> type[Block[Any, Any]] | None

Get the block class for a given block type.

Parameters:

Name	Type	Description	Default
block_type	str	The block type to look up	required

Returns:

Type	Description
type[Block[Any, Any]] | None	The registered block class, or None if not found

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

def get_block_class(self, block_type: str) -> type[Block[Any, Any]] | None:
    """Get the block class for a given block type.

    Args:
        block_type: The block type to look up

    Returns:
        The registered block class, or None if not found
    """
    return self._block_classes.get(block_type)

register

register(
    name: str,
    block_class: type[Block[Any, Any]],
    validators: list[ValidatorFunc] | None = None,
) -> None

Register a block class for a block type.

Parameters:

Name	Type	Description	Default
name	str	Block type name (e.g., "files_operations", "patch")	required
block_class	type[Block[Any, Any]]	Block class inheriting from Block[M, C]	required
validators	list[ValidatorFunc] | None	Optional list of validator functions for this block type	None

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

def register(
    self,
    name: str,
    block_class: type[Block[Any, Any]],
    validators: list[ValidatorFunc] | None = None,
) -> None:
    """Register a block class for a block type.

    Args:
        name: Block type name (e.g., "files_operations", "patch")
        block_class: Block class inheriting from Block[M, C]
        validators: Optional list of validator functions for this block type
    """
    self._block_classes[name] = block_class

    self.logger.debug(
        "block_type_registered",
        block_type=name,
        block_class=block_class.__name__,
        has_validators=validators is not None and len(validators) > 0,
    )

    # Add validators if provided
    if validators:
        for validator in validators:
            self.add_validator(name, validator)

register_template

register_template(
    version: str, template: str | Path, mode: str = "both"
) -> None

Register a custom prompt template for this registry.

Parameters:

Name	Type	Description	Default
version	str	Version identifier used by to_prompt(template_version=...).	required
template	str | Path	Template string, or a Path to a template file.	required
mode	str	"registry", "single", or "both".	'both'

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

def register_template(self, version: str, template: str | Path, mode: str = "both") -> None:
    """Register a custom prompt template for this registry.

    Args:
        version: Version identifier used by ``to_prompt(template_version=...)``.
        template: Template string, or a Path to a template file.
        mode: "registry", "single", or "both".
    """
    self._get_template_manager().register_template(version, template, mode)

serialize_block

serialize_block(block: Block[Any, Any]) -> str

Serialize a block instance using this registry's syntax.

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

def serialize_block(self, block: Block[Any, Any]) -> str:
    """Serialize a block instance using this registry's syntax."""
    return self._syntax.serialize_block(block)

to_prompt

to_prompt(
    *,
    include_examples: bool = True,
    template_version: str = "default",
) -> str

Generate an LLM system prompt documenting all registered blocks.

The prompt describes this registry's syntax format and, for each registered block type, its description, metadata fields, content format, and (optionally) serialized examples.

Parameters:

Name	Type	Description	Default
include_examples	bool	Whether to render each block's examples.	True
template_version	str	Template version for A/B testing.	'default'

Returns:

Type	Description
str	The rendered system prompt.

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

def to_prompt(self, *, include_examples: bool = True, template_version: str = "default") -> str:
    """Generate an LLM system prompt documenting all registered blocks.

    The prompt describes this registry's syntax format and, for each
    registered block type, its description, metadata fields, content
    format, and (optionally) serialized examples.

    Args:
        include_examples: Whether to render each block's examples.
        template_version: Template version for A/B testing.

    Returns:
        The rendered system prompt.
    """
    blocks = [
        build_block_context(block_class, self._syntax, include_examples=include_examples)
        for block_class in self._block_classes.values()
    ]
    context: dict[str, Any] = {
        "syntax_name": type(self._syntax).__name__,
        "syntax_format": self._syntax.describe_format(),
        "blocks": blocks,
    }
    return self._get_template_manager().render(context, template_version, mode="registry")

validate_block

validate_block(block: ExtractedBlock[Any, Any]) -> bool

Run all validators for a block.

Parameters:

Name	Type	Description	Default
block	ExtractedBlock[Any, Any]	Extracted block to validate	required

Returns:

Type	Description
bool	True if all validators pass

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

def validate_block(
    self,
    block: ExtractedBlock[Any, Any],
) -> bool:
    """Run all validators for a block.

    Args:
        block: Extracted block to validate

    Returns:
        True if all validators pass
    """
    # block.metadata is BaseMetadata which always has block_type
    block_type = block.metadata.block_type
    if not block_type:
        return True

    validators = self._validators.get(block_type, [])
    return all(v(block) for v in validators)

validate_content

validate_content(
    block_type: str,
    raw_content: str,
    parsed_content: dict[str, Any] | None,
) -> ValidationResult

Run all content validators for a block type.

Parameters:

Name	Type	Description	Default
block_type	str	Type of block being validated	required
raw_content	str	Raw content string	required
parsed_content	dict[str, Any] | None	Parsed content dict (if available)	required

Returns:

Type	Description
ValidationResult	ValidationResult with combined results from all validators

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

def validate_content(
    self,
    block_type: str,
    raw_content: str,
    parsed_content: dict[str, Any] | None,
) -> ValidationResult:
    """Run all content validators for a block type.

    Args:
        block_type: Type of block being validated
        raw_content: Raw content string
        parsed_content: Parsed content dict (if available)

    Returns:
        ValidationResult with combined results from all validators
    """
    validators = self._content_validators.get(block_type, [])
    if not validators:
        return ValidationResult.success()

    for validator in validators:
        result = validator(raw_content, parsed_content)
        if not result.passed:
            self.logger.debug(
                "content_validation_failed",
                block_type=block_type,
                validator_name=validator.__name__,
                error=result.error,
            )
            return result

    return ValidationResult.success()

validate_metadata

validate_metadata(
    block_type: str,
    raw_metadata: str,
    parsed_metadata: dict[str, Any] | None,
) -> ValidationResult

Run all metadata validators for a block type.

Parameters:

Name	Type	Description	Default
block_type	str	Type of block being validated	required
raw_metadata	str	Raw metadata string	required
parsed_metadata	dict[str, Any] | None	Parsed metadata dict (if available)	required

Returns:

Type	Description
ValidationResult	ValidationResult with combined results from all validators

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

def validate_metadata(
    self,
    block_type: str,
    raw_metadata: str,
    parsed_metadata: dict[str, Any] | None,
) -> ValidationResult:
    """Run all metadata validators for a block type.

    Args:
        block_type: Type of block being validated
        raw_metadata: Raw metadata string
        parsed_metadata: Parsed metadata dict (if available)

    Returns:
        ValidationResult with combined results from all validators
    """
    validators = self._metadata_validators.get(block_type, [])
    if not validators:
        return ValidationResult.success()

    for validator in validators:
        result = validator(raw_metadata, parsed_metadata)
        if not result.passed:
            self.logger.debug(
                "metadata_validation_failed",
                block_type=block_type,
                validator_name=validator.__name__,
                error=result.error,
            )
            return result

    return ValidationResult.success()

ValidationResult dataclass

Result from section validation.

Attributes:

Name	Type	Description
passed	bool	Whether validation succeeded
error	str | None	Error message if validation failed

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

@dataclass
class ValidationResult:
    """Result from section validation.

    Attributes:
        passed: Whether validation succeeded
        error: Error message if validation failed
    """

    passed: bool = True
    error: str | None = None

    @classmethod
    def success(cls) -> ValidationResult:
        """Create a successful validation result."""
        return cls(passed=True)

    @classmethod
    def failure(cls, error: str) -> ValidationResult:
        """Create a failed validation result."""
        return cls(passed=False, error=error)

error class-attribute instance-attribute

error: str | None = None

passed class-attribute instance-attribute

passed: bool = True

failure classmethod

failure(error: str) -> ValidationResult

Create a failed validation result.

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

@classmethod
def failure(cls, error: str) -> ValidationResult:
    """Create a failed validation result."""
    return cls(passed=False, error=error)

success classmethod

success() -> ValidationResult

Create a successful validation result.

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

@classmethod
def success(cls) -> ValidationResult:
    """Create a successful validation result."""
    return cls(passed=True)

hother.streamblocks.core.models

Core models for StreamBlocks.

Block

Bases: BaseModel

User-facing base class for defining block types.

This minimal class contains only the essential fields (metadata and content). Users inherit from this to define their block types.

Usage

class YesNo(Block[YesNoMetadata, YesNoContent]): pass

Access fields

block: Block[YesNoMetadata, YesNoContent] block.metadata.prompt # Type-safe access to metadata fields block.content.response # Type-safe access to content fields

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

class Block[TMetadata: BaseMetadata, TContent: BaseContent](BaseModel):
    """User-facing base class for defining block types.

    This minimal class contains only the essential fields (metadata and content).
    Users inherit from this to define their block types.

    Usage:
        class YesNo(Block[YesNoMetadata, YesNoContent]):
            pass

        # Access fields
        block: Block[YesNoMetadata, YesNoContent]
        block.metadata.prompt  # Type-safe access to metadata fields
        block.content.response  # Type-safe access to content fields
    """

    metadata: TMetadata = Field(..., description="Parsed block metadata")
    content: TContent = Field(..., description="Parsed block content")

    # Declared examples for this block type. Either a list of
    # {"metadata": {...}, "content": {...}} dicts, or a path (Path or str) to a
    # markdown file with YAML frontmatter that names the syntax. Used by prompt
    # generation.
    __examples__: ClassVar[list[dict[str, Any]] | Path | str] = []

    # Examples added at runtime via add_example(), keyed by concrete class so
    # subclasses do not share storage.
    _dynamic_examples: ClassVar[dict[type, list[Any]]] = {}
    # Cache of file-loaded examples keyed by (class, resolved_path, mtime).
    _examples_file_cache: ClassVar[dict[tuple[type, str, float], list[Any]]] = {}

    @classmethod
    def add_example(cls, example: Self | dict[str, Any]) -> None:
        """Add a single example for this block type.

        Args:
            example: A block instance, or a dict with ``metadata`` and
                ``content`` keys. Missing ``raw_content`` is auto-filled.
        """
        instance = cls._example_from_dict(example) if isinstance(example, dict) else example
        cls._dynamic_examples.setdefault(cls, []).append(instance)

    @classmethod
    def add_examples(cls, examples: list[Self | dict[str, Any]]) -> None:
        """Add multiple examples for this block type."""
        for example in examples:
            cls.add_example(example)

    @classmethod
    def clear_examples(cls) -> None:
        """Clear examples added via add_example().

        Does not affect examples declared in the ``__examples__`` attribute.
        """
        cls._dynamic_examples[cls] = []

    @classmethod
    def get_examples(cls) -> list[Self]:
        """Return all examples (declared + dynamically added) as instances.

        Declared examples come from ``__examples__`` (inline dicts or a markdown
        file). Dynamically added examples come from :meth:`add_example`.
        """
        declared = cls.__examples__
        if isinstance(declared, (str, Path)):
            examples = cls._load_examples_from_file(declared)
        else:
            examples = [cls._example_from_dict(item) for item in declared]
        examples.extend(cls._dynamic_examples.get(cls, []))
        return examples

    @classmethod
    def _example_from_dict(cls, data: dict[str, Any]) -> Self:
        """Validate an example dict into an instance, auto-filling raw_content."""
        return cls.model_validate(cls._ensure_raw_content(data))

    @classmethod
    def _ensure_raw_content(cls, data: dict[str, Any]) -> dict[str, Any]:
        """Fill a missing content ``raw_content`` by serializing content fields.

        Honors an explicit ``raw_content`` when present. The serialization
        format follows the block's declared content format
        (``__content_format__`` set by ``@parse_as_json`` / ``@parse_as_yaml``),
        defaulting to JSON. Block types with bespoke content formats should
        provide ``raw_content`` explicitly in their examples.
        """
        content_value = data.get("content")
        if not isinstance(content_value, dict) or "raw_content" in content_value:
            return data
        content = cast("dict[str, Any]", content_value)
        fields = {key: value for key, value in content.items() if key != "raw_content"}
        _, content_class = extract_block_types(cls)
        if content_class.__content_format__ == "yaml":
            raw_content = yaml.dump(fields, default_flow_style=False, sort_keys=False).strip()
        else:
            raw_content = json.dumps(fields, ensure_ascii=False)
        return {**data, "content": {**content, "raw_content": raw_content}}

    @classmethod
    def _load_examples_from_file(cls, path: Path | str) -> list[Self]:
        """Load and cache examples from a markdown file with YAML frontmatter."""
        file_path = Path(path).resolve()
        if not file_path.is_file():
            msg = f"Examples file not found: {file_path}"
            raise FileNotFoundError(msg)

        mtime = file_path.stat().st_mtime
        cache_key = (cls, str(file_path), mtime)
        cached = cls._examples_file_cache.get(cache_key)
        if cached is not None:
            return list(cached)

        content = file_path.read_text(encoding="utf-8")
        syntax, blocks_content = cls._parse_examples_frontmatter(content, file_path)
        examples = cls._extract_blocks_from_content(blocks_content, syntax)
        cls._examples_file_cache[cache_key] = examples
        return list(examples)

    @classmethod
    def _parse_examples_frontmatter(cls, content: str, file_path: Path) -> tuple[BaseSyntax, str]:
        """Split a markdown examples file into (syntax instance, blocks text)."""
        from hother.streamblocks.syntaxes.factory import get_syntax_instance
        from hother.streamblocks.syntaxes.models import Syntax

        if not content.lstrip().startswith(_FRONTMATTER_FENCE):
            msg = f"Examples file must start with YAML frontmatter: {file_path}"
            raise ValueError(msg)

        parts = content.split(_FRONTMATTER_FENCE, 2)
        if len(parts) < _FRONTMATTER_PART_COUNT:
            msg = f"Invalid frontmatter in examples file: {file_path}"
            raise ValueError(msg)

        parsed = yaml.safe_load(parts[1].strip())
        if not isinstance(parsed, dict):
            msg = f"Frontmatter must be a YAML mapping: {file_path}"
            raise TypeError(msg)

        frontmatter = cast("dict[str, Any]", parsed)
        syntax_name = frontmatter.get("syntax")
        if not isinstance(syntax_name, str):
            msg = f"Frontmatter must name a string 'syntax' field: {file_path}"
            raise TypeError(msg)

        try:
            syntax = get_syntax_instance(Syntax[syntax_name])
        except KeyError as error:
            valid = ", ".join(member.name for member in Syntax)
            msg = f"Unknown syntax '{syntax_name}' in {file_path}. Valid options: {valid}"
            raise ValueError(msg) from error

        return syntax, parts[2].strip()

    @classmethod
    def _extract_blocks_from_content(cls, content: str, syntax: BaseSyntax) -> list[Self]:
        """Parse all blocks in a text body using the given syntax."""
        examples: list[Self] = []
        candidate: BlockCandidate | None = None
        for line in content.split("\n"):
            if candidate is None:
                if syntax.detect_line(line, None).is_opening:
                    candidate = BlockCandidate(syntax=syntax, start_line=0)
                    candidate.add_line(line)
                continue
            candidate.add_line(line)
            if syntax.detect_line(line, candidate).is_closing:
                result = syntax.parse_block(candidate, block_class=cls)
                if not result.success or result.metadata is None or result.content is None:
                    msg = f"Failed to parse example block: {result.error or 'incomplete block'}"
                    raise ValueError(msg)
                examples.append(cls.model_validate({"metadata": result.metadata, "content": result.content}))
                candidate = None
        return examples

content class-attribute instance-attribute

content: TContent = Field(
    ..., description="Parsed block content"
)

metadata class-attribute instance-attribute

metadata: TMetadata = Field(
    ..., description="Parsed block metadata"
)

add_example classmethod

add_example(example: Self | dict[str, Any]) -> None

Add a single example for this block type.

Parameters:

Name	Type	Description	Default
example	Self | dict[str, Any]	A block instance, or a dict with metadata and content keys. Missing raw_content is auto-filled.	required

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

@classmethod
def add_example(cls, example: Self | dict[str, Any]) -> None:
    """Add a single example for this block type.

    Args:
        example: A block instance, or a dict with ``metadata`` and
            ``content`` keys. Missing ``raw_content`` is auto-filled.
    """
    instance = cls._example_from_dict(example) if isinstance(example, dict) else example
    cls._dynamic_examples.setdefault(cls, []).append(instance)

add_examples classmethod

add_examples(examples: list[Self | dict[str, Any]]) -> None

Add multiple examples for this block type.

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

@classmethod
def add_examples(cls, examples: list[Self | dict[str, Any]]) -> None:
    """Add multiple examples for this block type."""
    for example in examples:
        cls.add_example(example)

clear_examples classmethod

clear_examples() -> None

Clear examples added via add_example().

Does not affect examples declared in the __examples__ attribute.

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

@classmethod
def clear_examples(cls) -> None:
    """Clear examples added via add_example().

    Does not affect examples declared in the ``__examples__`` attribute.
    """
    cls._dynamic_examples[cls] = []

get_examples classmethod

get_examples() -> list[Self]

Return all examples (declared + dynamically added) as instances.

Declared examples come from __examples__ (inline dicts or a markdown file). Dynamically added examples come from :meth:add_example.

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

@classmethod
def get_examples(cls) -> list[Self]:
    """Return all examples (declared + dynamically added) as instances.

    Declared examples come from ``__examples__`` (inline dicts or a markdown
    file). Dynamically added examples come from :meth:`add_example`.
    """
    declared = cls.__examples__
    if isinstance(declared, (str, Path)):
        examples = cls._load_examples_from_file(declared)
    else:
        examples = [cls._example_from_dict(item) for item in declared]
    examples.extend(cls._dynamic_examples.get(cls, []))
    return examples

BlockCandidate

Tracks a potential block being accumulated.

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

class BlockCandidate:
    """Tracks a potential block being accumulated."""

    __slots__ = (
        "content_lines",
        "content_validation_error",
        "content_validation_passed",
        "current_section",
        "lines",
        "metadata_lines",
        "metadata_validation_error",
        "metadata_validation_passed",
        "parsed_content",
        "parsed_metadata",
        "start_line",
        "state",
        "syntax",
    )

    def __init__(self, syntax: BaseSyntax, start_line: int) -> None:
        """Initialize a new block candidate.

        Args:
            syntax: The syntax handler for this block
            start_line: Line number where the block started
        """
        self.syntax = syntax
        self.start_line = start_line
        self.lines: list[str] = []
        self.state = BlockState.HEADER_DETECTED
        self.metadata_lines: list[str] = []
        self.content_lines: list[str] = []
        self.current_section: SectionType = SectionType.HEADER

        # Cache fields for early parsing results
        self.parsed_metadata: dict[str, Any] | None = None
        self.parsed_content: dict[str, Any] | None = None

        # Validation state for section end events
        self.metadata_validation_passed: bool = True
        self.metadata_validation_error: str | None = None
        self.content_validation_passed: bool = True
        self.content_validation_error: str | None = None

    def add_line(self, line: str) -> None:
        """Add a line to the candidate."""
        self.lines.append(line)

    def transition_to_metadata(self) -> None:
        """Transition from header to metadata section.

        This method encapsulates the section state transition logic,
        making the state change explicit and centralized.
        """
        self.current_section = SectionType.METADATA

    def transition_to_content(self) -> None:
        """Transition from metadata/header to content section.

        This method encapsulates the section state transition logic,
        making the state change explicit and centralized.
        """
        self.current_section = SectionType.CONTENT

    def cache_metadata_validation(self, passed: bool, error: str | None) -> None:
        """Cache metadata validation result.

        This method encapsulates validation result storage, providing
        a clear interface for the state machine to cache validation state.

        Args:
            passed: Whether metadata validation passed
            error: Error message if validation failed, None otherwise
        """
        self.metadata_validation_passed = passed
        self.metadata_validation_error = error

    def cache_content_validation(self, passed: bool, error: str | None) -> None:
        """Cache content validation result.

        This method encapsulates validation result storage, providing
        a clear interface for the state machine to cache validation state.

        Args:
            passed: Whether content validation passed
            error: Error message if validation failed, None otherwise
        """
        self.content_validation_passed = passed
        self.content_validation_error = error

    @property
    def raw_text(self) -> str:
        """Get the raw text of all accumulated lines."""
        return "\n".join(self.lines)

    def compute_hash(self) -> str:
        """Compute hash of first N chars for ID (N defined in constants)."""
        text_slice = self.raw_text[: LIMITS.HASH_PREFIX_LENGTH]
        return hashlib.sha256(text_slice.encode()).hexdigest()[:8]

    def __repr__(self) -> str:
        """Return a developer-friendly string representation."""
        return (
            f"BlockCandidate(syntax={type(self.syntax).__name__}, "
            f"start_line={self.start_line}, state={self.state.value}, "
            f"lines={len(self.lines)}, section={self.current_section!r})"
        )

content_lines instance-attribute

content_lines: list[str] = []

content_validation_error instance-attribute

content_validation_error: str | None = None

content_validation_passed instance-attribute

content_validation_passed: bool = True

current_section instance-attribute

current_section: SectionType = HEADER

lines instance-attribute

lines: list[str] = []

metadata_lines instance-attribute

metadata_lines: list[str] = []

metadata_validation_error instance-attribute

metadata_validation_error: str | None = None

metadata_validation_passed instance-attribute

metadata_validation_passed: bool = True

parsed_content instance-attribute

parsed_content: dict[str, Any] | None = None

parsed_metadata instance-attribute

parsed_metadata: dict[str, Any] | None = None

raw_text property

raw_text: str

Get the raw text of all accumulated lines.

start_line instance-attribute

start_line = start_line

state instance-attribute

state = HEADER_DETECTED

syntax instance-attribute

syntax = syntax

add_line

add_line(line: str) -> None

Add a line to the candidate.

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

def add_line(self, line: str) -> None:
    """Add a line to the candidate."""
    self.lines.append(line)

cache_content_validation

cache_content_validation(
    passed: bool, error: str | None
) -> None

Cache content validation result.

This method encapsulates validation result storage, providing a clear interface for the state machine to cache validation state.

Parameters:

Name	Type	Description	Default
passed	bool	Whether content validation passed	required
error	str | None	Error message if validation failed, None otherwise	required

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

def cache_content_validation(self, passed: bool, error: str | None) -> None:
    """Cache content validation result.

    This method encapsulates validation result storage, providing
    a clear interface for the state machine to cache validation state.

    Args:
        passed: Whether content validation passed
        error: Error message if validation failed, None otherwise
    """
    self.content_validation_passed = passed
    self.content_validation_error = error

cache_metadata_validation

cache_metadata_validation(
    passed: bool, error: str | None
) -> None

Cache metadata validation result.

This method encapsulates validation result storage, providing a clear interface for the state machine to cache validation state.

Parameters:

Name	Type	Description	Default
passed	bool	Whether metadata validation passed	required
error	str | None	Error message if validation failed, None otherwise	required

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

def cache_metadata_validation(self, passed: bool, error: str | None) -> None:
    """Cache metadata validation result.

    This method encapsulates validation result storage, providing
    a clear interface for the state machine to cache validation state.

    Args:
        passed: Whether metadata validation passed
        error: Error message if validation failed, None otherwise
    """
    self.metadata_validation_passed = passed
    self.metadata_validation_error = error

compute_hash

compute_hash() -> str

Compute hash of first N chars for ID (N defined in constants).

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

def compute_hash(self) -> str:
    """Compute hash of first N chars for ID (N defined in constants)."""
    text_slice = self.raw_text[: LIMITS.HASH_PREFIX_LENGTH]
    return hashlib.sha256(text_slice.encode()).hexdigest()[:8]

transition_to_content

transition_to_content() -> None

Transition from metadata/header to content section.

This method encapsulates the section state transition logic, making the state change explicit and centralized.

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

def transition_to_content(self) -> None:
    """Transition from metadata/header to content section.

    This method encapsulates the section state transition logic,
    making the state change explicit and centralized.
    """
    self.current_section = SectionType.CONTENT

transition_to_metadata

transition_to_metadata() -> None

Transition from header to metadata section.

This method encapsulates the section state transition logic, making the state change explicit and centralized.

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

def transition_to_metadata(self) -> None:
    """Transition from header to metadata section.

    This method encapsulates the section state transition logic,
    making the state change explicit and centralized.
    """
    self.current_section = SectionType.METADATA

ExtractedBlock

Bases: Block[TMetadata, TContent]

Full runtime representation of an extracted block.

This class extends the minimal Block with extraction metadata like line numbers, syntax name, and hash ID. The processor creates instances of this class when blocks are successfully extracted.

The metadata and content fields are typed generics, allowing type-safe access to block-specific fields.

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

class ExtractedBlock[TMetadata: BaseMetadata, TContent: BaseContent](Block[TMetadata, TContent]):
    """Full runtime representation of an extracted block.

    This class extends the minimal Block with extraction metadata like
    line numbers, syntax name, and hash ID. The processor creates instances
    of this class when blocks are successfully extracted.

    The metadata and content fields are typed generics, allowing type-safe
    access to block-specific fields.
    """

    syntax_name: str = Field(..., description="Name of the syntax that extracted this block")
    raw_text: str = Field(..., description="Original raw text of the block")
    line_start: int = Field(..., description="Starting line number")
    line_end: int = Field(..., description="Ending line number")
    hash_id: str = Field(..., description="Hash-based ID for the block")

content class-attribute instance-attribute

content: TContent = Field(
    ..., description="Parsed block content"
)

hash_id class-attribute instance-attribute

hash_id: str = Field(
    ..., description="Hash-based ID for the block"
)

line_end class-attribute instance-attribute

line_end: int = Field(..., description="Ending line number")

line_start class-attribute instance-attribute

line_start: int = Field(
    ..., description="Starting line number"
)

metadata class-attribute instance-attribute

metadata: TMetadata = Field(
    ..., description="Parsed block metadata"
)

raw_text class-attribute instance-attribute

raw_text: str = Field(
    ..., description="Original raw text of the block"
)

syntax_name class-attribute instance-attribute

syntax_name: str = Field(
    ...,
    description="Name of the syntax that extracted this block",
)

add_example classmethod

add_example(example: Self | dict[str, Any]) -> None

Add a single example for this block type.

Parameters:

Name	Type	Description	Default
example	Self | dict[str, Any]	A block instance, or a dict with metadata and content keys. Missing raw_content is auto-filled.	required

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

@classmethod
def add_example(cls, example: Self | dict[str, Any]) -> None:
    """Add a single example for this block type.

    Args:
        example: A block instance, or a dict with ``metadata`` and
            ``content`` keys. Missing ``raw_content`` is auto-filled.
    """
    instance = cls._example_from_dict(example) if isinstance(example, dict) else example
    cls._dynamic_examples.setdefault(cls, []).append(instance)

add_examples classmethod

add_examples(examples: list[Self | dict[str, Any]]) -> None

Add multiple examples for this block type.

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

@classmethod
def add_examples(cls, examples: list[Self | dict[str, Any]]) -> None:
    """Add multiple examples for this block type."""
    for example in examples:
        cls.add_example(example)

clear_examples classmethod

clear_examples() -> None

Clear examples added via add_example().

Does not affect examples declared in the __examples__ attribute.

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

@classmethod
def clear_examples(cls) -> None:
    """Clear examples added via add_example().

    Does not affect examples declared in the ``__examples__`` attribute.
    """
    cls._dynamic_examples[cls] = []

get_examples classmethod

get_examples() -> list[Self]

Return all examples (declared + dynamically added) as instances.

Declared examples come from __examples__ (inline dicts or a markdown file). Dynamically added examples come from :meth:add_example.

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

@classmethod
def get_examples(cls) -> list[Self]:
    """Return all examples (declared + dynamically added) as instances.

    Declared examples come from ``__examples__`` (inline dicts or a markdown
    file). Dynamically added examples come from :meth:`add_example`.
    """
    declared = cls.__examples__
    if isinstance(declared, (str, Path)):
        examples = cls._load_examples_from_file(declared)
    else:
        examples = [cls._example_from_dict(item) for item in declared]
    examples.extend(cls._dynamic_examples.get(cls, []))
    return examples

extract_block_types

extract_block_types(
    block_class: type[Any],
) -> tuple[type[BaseMetadata], type[BaseContent]]

Extract metadata and content type parameters from a Block class.

For classes inheriting from Block[M, C], Pydantic resolves the concrete types and stores them in the field annotations during class definition. This function extracts those resolved types from model_fields.

Parameters:

Name	Type	Description	Default
block_class	type[Any]	The block class to extract types from	required

Returns:

Type	Description
tuple[type[BaseMetadata], type[BaseContent]]	Tuple of (metadata_class, content_class)

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

def extract_block_types(block_class: type[Any]) -> tuple[type[BaseMetadata], type[BaseContent]]:
    """Extract metadata and content type parameters from a Block class.

    For classes inheriting from Block[M, C], Pydantic resolves the concrete
    types and stores them in the field annotations during class definition.
    This function extracts those resolved types from model_fields.

    Args:
        block_class: The block class to extract types from

    Returns:
        Tuple of (metadata_class, content_class)
    """
    # Extract type parameters from Pydantic field annotations
    # Pydantic resolves Block[M, C] generics during class definition
    if issubclass(block_class, BaseModel):
        metadata_field = block_class.model_fields.get("metadata")
        content_field = block_class.model_fields.get("content")

        if (
            metadata_field
            and content_field
            and metadata_field.annotation is not None
            and content_field.annotation is not None
        ):
            return (metadata_field.annotation, content_field.annotation)

    # Fallback to base classes
    return (BaseMetadata, BaseContent)