Validation

Parsing guarantees structure; validators enforce your rules on top of it. The registry supports three validator kinds, each running at a different point in the block lifecycle.

Three validation stages

Stage	Registered with	Signature	Runs when
Metadata	registry.add_metadata_validator(block_type, fn)	(raw: str, parsed: dict \| None) -> ValidationResult	Metadata section completes, before content accumulates
Content	registry.add_content_validator(block_type, fn)	(raw: str, parsed: dict \| None) -> ValidationResult	Content section completes, before the final BlockEndEvent
Block	registry.add_validator(block_type, fn) or register(..., validators=[fn])	(block: ExtractedBlock) -> bool	After the full block is parsed

Validators of each kind run in registration order; the first failure stops the chain.

Early metadata validation

Metadata validators receive the raw metadata string and the parsed dict (or None if parsing failed) and return a ValidationResult:

def validate_required_fields(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
    """Validate that required metadata fields are present."""
    if not parsed:
        return ValidationResult.failure("No metadata parsed")
    if "id" not in parsed:
        return ValidationResult.failure("Missing required field: id")
    return ValidationResult.success()


def validate_id_format(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
    """Validate that ID follows naming convention."""
    if not parsed:
        return ValidationResult.success()  # Let other validator handle this
    block_id = parsed.get("id", "")
    if not block_id.startswith("ops-"):
        return ValidationResult.failure(f"ID must start with 'ops-', got: {block_id}")
    return ValidationResult.success()

ValidationResult is a small dataclass with two constructors:

ValidationResult.success()
ValidationResult.failure("ID must start with 'ops-'")

Because metadata validators fire at the end of the metadata section, you can reject a block before its content streams in, useful for skipping large payloads you already know you don't want.

Failure modes

What happens after a metadata validation failure is controlled by the registry's metadata_failure_mode:

Mode	Behavior
MetadataValidationFailureMode.ABORT_BLOCK (default)	Emit a BlockErrorEvent immediately and stop processing the block
MetadataValidationFailureMode.CONTINUE	Log a warning and keep processing content normally
MetadataValidationFailureMode.SKIP_CONTENT	Skip content accumulation and emit a partial block

# Mode 1: ABORT_BLOCK (default) - stop processing on validation failure
registry = Registry(
    syntax=syntax,
    metadata_failure_mode=MetadataValidationFailureMode.ABORT_BLOCK,
)
registry.register("files_operations", FileOperations)
registry.add_metadata_validator("files_operations", validate_id_format)
processor = StreamBlockProcessor(registry)

# This block has an invalid ID (doesn't start with 'ops-')
text = dedent("""
    !!start
    ---
    id: invalid-id
    block_type: files_operations
    ---
    src/main.py:C
    !!end
""").strip()

print("=== ABORT_BLOCK mode ===")
async for event in processor.process_stream(simple_text_stream(text)):
    if isinstance(event, BlockErrorEvent):
        print(f"Error: {event.reason}")
    elif isinstance(event, BlockEndEvent):
        print(f"Block extracted: {event.get_block()}")

Content and block validators

Content validators mirror metadata validators but run on the raw content when the content section ends. Block validators are plain predicates over the final ExtractedBlock; returning False rejects the block with a BlockErrorEvent (VALIDATION_FAILED).

Composing validators

All three kinds combine on a single block type:

def validate_has_description(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
    """First metadata validator: check for description."""
    if parsed and "description" not in parsed:
        return ValidationResult.failure("Missing 'description' in metadata")
    return ValidationResult.success()


def validate_id_length(raw: str, parsed: dict[str, Any] | None) -> ValidationResult:
    """Second metadata validator: check ID length."""
    min_id_length = 3
    if parsed:
        block_id = str(parsed.get("id", ""))
        if len(block_id) < min_id_length:
            return ValidationResult.failure(f"ID too short: {len(block_id)} < {min_id_length}")
    return ValidationResult.success()

def validate_max_operations(block: ExtractedBlock[Any, Any]) -> bool:
    """General validator: limit number of operations."""
    max_ops = 5
    if hasattr(block.content, "operations"):
        return len(block.content.operations) <= max_ops
    return True

# Register metadata validators (run in order, first failure stops)
registry.add_metadata_validator("files_operations", validate_has_description)
registry.add_metadata_validator("files_operations", validate_id_length)

# Register content validators (run in order, first failure stops)
registry.add_content_validator("files_operations", validate_has_operations)
registry.add_content_validator("files_operations", validate_no_delete)

# Register general validator (runs after full block extraction)
registry.add_validator("files_operations", validate_max_operations)

A block must pass every registered validator to be delivered through BlockEndEvent.

Next steps

• Error Handling: react to BlockErrorEvent and its error codes.
• Defining Custom Blocks: parsing, the stage before validation.
• Events: BlockMetadataEndEvent and BlockContentEndEvent carry the validation outcome.