Schema Validators

akms.schema.validators

YAML frontmatter parsers and validators for AKMS v2 schemas.

All parse functions validate schema version and return typed Pydantic models.

parse_node_frontmatter

parse_node_frontmatter(
    path: str | Path,
    *,
    is_local: bool = False,
    is_code_mirror: bool = False,
) -> (
    GlobalNodeFrontmatter
    | LocalNodeFrontmatter
    | CodeMirrorNodeFrontmatter
)

Parse and validate a knowledge node's YAML frontmatter.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the .md node file.	required
is_local	bool	If True, validates as a local node (allows agent source).	False
is_code_mirror	bool	If True, validates as a code-mirror node.	False

Returns:

Type	Description
GlobalNodeFrontmatter | LocalNodeFrontmatter | CodeMirrorNodeFrontmatter	Validated Pydantic model.

Raises:

Type	Description
SchemaVersionError	If akms_schema doesn't match.
SchemaValidationError	If required fields are missing or invalid.
FileNotFoundError	If file doesn't exist.

Source code in packages/akms/src/akms/schema/validators.py

def parse_node_frontmatter(
    path: str | Path,
    *,
    is_local: bool = False,
    is_code_mirror: bool = False,
) -> GlobalNodeFrontmatter | LocalNodeFrontmatter | CodeMirrorNodeFrontmatter:
    """Parse and validate a knowledge node's YAML frontmatter.

    Args:
        path: Path to the .md node file.
        is_local: If True, validates as a local node (allows agent source).
        is_code_mirror: If True, validates as a code-mirror node.

    Returns:
        Validated Pydantic model.

    Raises:
        SchemaVersionError: If akms_schema doesn't match.
        SchemaValidationError: If required fields are missing or invalid.
        FileNotFoundError: If file doesn't exist.
    """
    path_str = str(path)
    data, _content = _load_frontmatter(path)
    _check_schema_version(data, path_str)

    try:
        if is_code_mirror:
            return CodeMirrorNodeFrontmatter(**data)
        elif is_local:
            return LocalNodeFrontmatter(**data)
        else:
            # Global node: reject experiential fields
            _check_no_experiential_fields(data, path_str)
            return GlobalNodeFrontmatter(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path_str) from e

parse_node_frontmatter_from_dict

parse_node_frontmatter_from_dict(
    data: dict[str, Any],
    *,
    is_local: bool = False,
    is_code_mirror: bool = False,
    path: str | None = None,
) -> (
    GlobalNodeFrontmatter
    | LocalNodeFrontmatter
    | CodeMirrorNodeFrontmatter
)

Parse and validate frontmatter from a pre-loaded dict.

Useful when frontmatter has already been extracted (e.g., during build).

Source code in packages/akms/src/akms/schema/validators.py

def parse_node_frontmatter_from_dict(
    data: dict[str, Any],
    *,
    is_local: bool = False,
    is_code_mirror: bool = False,
    path: str | None = None,
) -> GlobalNodeFrontmatter | LocalNodeFrontmatter | CodeMirrorNodeFrontmatter:
    """Parse and validate frontmatter from a pre-loaded dict.

    Useful when frontmatter has already been extracted (e.g., during build).
    """
    _check_schema_version(data, path)

    try:
        if is_code_mirror:
            return CodeMirrorNodeFrontmatter(**data)
        elif is_local:
            return LocalNodeFrontmatter(**data)
        else:
            _check_no_experiential_fields(data, path)
            return GlobalNodeFrontmatter(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path) from e

parse_local_state

parse_local_state(path: str | Path) -> LocalStateOverlay

Parse and validate local_state.yaml.

Returns:

Type	Description
LocalStateOverlay	Validated LocalStateOverlay model.

Source code in packages/akms/src/akms/schema/validators.py

def parse_local_state(path: str | Path) -> LocalStateOverlay:
    """Parse and validate local_state.yaml.

    Returns:
        Validated LocalStateOverlay model.
    """
    path_str = str(path)
    data = _load_yaml(path)

    if not data:
        # Empty file → return default overlay
        return LocalStateOverlay()

    _check_schema_version(data, path_str)

    try:
        return LocalStateOverlay(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path_str) from e

parse_agent_memory

parse_agent_memory(path: str | Path) -> AgentMemory

Parse and validate a per-task AgentMemory file.

Returns:

Type	Description
AgentMemory	Validated AgentMemory model.

Source code in packages/akms/src/akms/schema/validators.py

def parse_agent_memory(path: str | Path) -> AgentMemory:
    """Parse and validate a per-task AgentMemory file.

    Returns:
        Validated AgentMemory model.
    """
    path_str = str(path)
    data, _content = _load_frontmatter(path)
    _check_schema_version(data, path_str)

    try:
        return AgentMemory(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path_str) from e

parse_pcd

parse_pcd(path: str | Path) -> PCD

Parse and validate a Phase Completion Document.

Returns:

Type	Description
PCD	Validated PCD model with zone extraction methods.

Source code in packages/akms/src/akms/schema/validators.py

def parse_pcd(path: str | Path) -> PCD:
    """Parse and validate a Phase Completion Document.

    Returns:
        Validated PCD model with zone extraction methods.
    """
    path_str = str(path)
    data, _content = _load_frontmatter(path)
    _check_schema_version(data, path_str)

    try:
        return PCD(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path_str) from e

parse_propagation_config

parse_propagation_config(
    path: str | Path,
) -> PropagationConfig

Parse and validate propagation_config.yaml.

Returns:

Type	Description
PropagationConfig	Validated PropagationConfig model.

Source code in packages/akms/src/akms/schema/validators.py

def parse_propagation_config(path: str | Path) -> PropagationConfig:
    """Parse and validate propagation_config.yaml.

    Returns:
        Validated PropagationConfig model.
    """
    path_str = str(path)
    data = _load_yaml(path)

    if not data:
        return PropagationConfig()

    _check_schema_version(data, path_str)

    try:
        return PropagationConfig(**data)
    except ValidationError as e:
        raise SchemaValidationError(str(e), path_str) from e