Index

validation

Enhanced validation module for Marianne job configurations.

This module provides comprehensive validation beyond Pydantic schema validation, including Jinja template syntax checking, path existence validation, regex pattern compilation, and cross-reference checks.

The validation system is designed for two use cases: 1. Pre-execution validation via mzt validate command 2. Self-healing diagnosis to identify fixable configuration issues

Example usage

from marianne.validation import ValidationRunner, create_default_checks

runner = ValidationRunner(create_default_checks()) issues = runner.validate(config, config_path, raw_yaml)

for issue in issues: print(f"[{issue.severity}] {issue.check_id}: {issue.message}")

Classes

ValidationCheck

Bases: Protocol

Protocol for configuration validation checks.

Each check examines a specific aspect of the configuration and returns a list of ValidationIssue objects (empty if check passes).

Checks should be: - Idempotent (safe to run multiple times) - Side-effect free (don't modify anything) - Fast (validation should complete quickly)

Attributes

check_id property

check_id

Unique identifier for this check (e.g., V001).

severity property

severity

Default severity for issues from this check.

description property

description

Human-readable description of what this check does.

Functions

check

check(config, config_path, raw_yaml)

Run the validation check.

Parameters:

Name	Type	Description	Default
config	JobConfig	Parsed JobConfig object (already validated by Pydantic)	required
config_path	Path	Path to the config file (for resolving relative paths)	required
raw_yaml	str	Raw YAML text (for line number extraction)	required

Returns:

Type	Description
list[ValidationIssue]	List of ValidationIssue objects (empty if check passes)

Source code in src/marianne/validation/base.py

def check(
    self,
    config: "JobConfig",
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Run the validation check.

    Args:
        config: Parsed JobConfig object (already validated by Pydantic)
        config_path: Path to the config file (for resolving relative paths)
        raw_yaml: Raw YAML text (for line number extraction)

    Returns:
        List of ValidationIssue objects (empty if check passes)
    """
    ...

ValidationIssue dataclass

ValidationIssue(check_id, severity, message, line=None, column=None, context=None, suggestion=None, auto_fixable=False, metadata=dict())

A single validation issue found in the configuration.

Attributes:

Name	Type	Description
check_id	str	Unique identifier for the check (e.g., V001, V101)
severity	ValidationSeverity	ERROR, WARNING, or INFO
message	str	Human-readable description of the issue
line	int | None	Line number in config file (if applicable)
column	int | None	Column number (if applicable)
context	str | None	Surrounding text for context
suggestion	str | None	How to fix the issue
auto_fixable	bool	Whether --self-healing can automatically fix this

Functions

format_short

format_short()

Format as a single-line summary.

Source code in src/marianne/validation/base.py

def format_short(self) -> str:
    """Format as a single-line summary."""
    loc = f"Line {self.line}: " if self.line else ""
    return f"[{self.check_id}] {loc}{self.message}"

format_full

format_full()

Format with full details including suggestion.

Source code in src/marianne/validation/base.py

def format_full(self) -> str:
    """Format with full details including suggestion."""
    lines = [self.format_short()]
    if self.context:
        lines.append(f"         Context: {self.context}")
    if self.suggestion:
        lines.append(f"         Suggestion: {self.suggestion}")
    return "\n".join(lines)

ValidationSeverity

Bases: str, Enum

Severity level for validation issues.

• ERROR: Must fix before execution - job will fail
• WARNING: Should fix - may cause runtime failures
• INFO: Informational - consider reviewing

ValidationReporter

ValidationReporter(console=None)

Formats and outputs validation results.

Supports multiple output formats: - Terminal output with colors (default) - JSON for machine parsing - Plain text for logs

Initialize reporter.

Parameters:

Name	Type	Description	Default
console	Console | None	Rich Console for output. Creates one if not provided.	None

Source code in src/marianne/validation/reporter.py

def __init__(self, console: Console | None = None):
    """Initialize reporter.

    Args:
        console: Rich Console for output. Creates one if not provided.
    """
    self.console = console or Console()

Functions

report_terminal

report_terminal(issues, config_name, show_passed=True)

Output validation results to terminal with formatting.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue]	List of validation issues	required
config_name	str	Name of the config being validated	required
show_passed	bool	Whether to show passed checks summary	True

Source code in src/marianne/validation/reporter.py

def report_terminal(
    self,
    issues: list[ValidationIssue],
    config_name: str,
    show_passed: bool = True,
) -> None:
    """Output validation results to terminal with formatting.

    Args:
        issues: List of validation issues
        config_name: Name of the config being validated
        show_passed: Whether to show passed checks summary
    """
    self.console.print()

    # Count by severity
    errors = [i for i in issues if i.severity == ValidationSeverity.ERROR]
    warnings = [i for i in issues if i.severity == ValidationSeverity.WARNING]
    infos = [i for i in issues if i.severity == ValidationSeverity.INFO]

    # Print header based on results
    if not issues:
        self.console.print(
            Panel(
                f"[green]✓ Configuration valid:[/green] {config_name}",
                border_style="green",
            )
        )
        return

    # Print issues by severity
    if errors:
        self._print_section("ERRORS (must fix before running)", errors, "red")

    if warnings:
        self._print_section("WARNINGS (may cause issues)", warnings, "yellow")

    if infos:
        self._print_section("INFO (consider reviewing)", infos, "blue")

    # Print summary
    self.console.print()
    summary_parts = []
    if errors:
        err_s = "s" if len(errors) > 1 else ""
        summary_parts.append(
            f"[red]{len(errors)} error{err_s} (must fix)[/red]"
        )
    if warnings:
        warn_s = "s" if len(warnings) > 1 else ""
        summary_parts.append(
            f"[yellow]{len(warnings)} warning{warn_s}"
            f" (should fix)[/yellow]"
        )
    if infos:
        info_s = "s" if len(infos) > 1 else ""
        summary_parts.append(
            f"[blue]{len(infos)} info note{info_s}[/blue]"
        )

    self.console.print(f"Summary: {', '.join(summary_parts)}")

    # Final status
    if errors:
        self.console.print("\n[bold red]Validation: FAILED[/bold red]")
    else:
        self.console.print("\n[bold green]Validation: PASSED[/bold green] (with warnings)")

report_json

report_json(issues)

Output validation results as JSON.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue]	List of validation issues	required

Returns:

Type	Description
str	JSON string representation

Source code in src/marianne/validation/reporter.py

def report_json(self, issues: list[ValidationIssue]) -> str:
    """Output validation results as JSON.

    Args:
        issues: List of validation issues

    Returns:
        JSON string representation
    """
    result: dict[str, Any] = {
        "valid": not any(i.severity == ValidationSeverity.ERROR for i in issues),
        "error_count": sum(1 for i in issues if i.severity == ValidationSeverity.ERROR),
        "warning_count": sum(1 for i in issues if i.severity == ValidationSeverity.WARNING),
        "info_count": sum(1 for i in issues if i.severity == ValidationSeverity.INFO),
        "issues": [self._issue_to_dict(i) for i in issues],
    }
    return json.dumps(result, indent=2)

report_rendering_terminal

report_rendering_terminal(preview, verbose=False)

Output a rendering preview to the terminal.

Parameters:

Name	Type	Description	Default
preview	RenderingPreview	The rendering preview to display.	required
verbose	bool	If False, show sheet 1 only. If True, show all sheets.	False

Source code in src/marianne/validation/reporter.py

def report_rendering_terminal(
    self,
    preview: "RenderingPreview",
    verbose: bool = False,
) -> None:
    """Output a rendering preview to the terminal.

    Args:
        preview: The rendering preview to display.
        verbose: If False, show sheet 1 only. If True, show all sheets.
    """
    sheets_to_show = preview.sheets if verbose else preview.sheets[:1]

    for sp in sheets_to_show:
        header = f"Rendering Preview (Sheet {sp.sheet_num} of {preview.total_sheets})"
        if verbose and (sp.stage is not None or sp.instance is not None):
            parts: list[str] = []
            if sp.stage is not None:
                parts.append(f"stage={sp.stage}")
            if sp.instance is not None:
                parts.append(f"instance={sp.instance}")
            header += f" \\[{', '.join(parts)}]"

        self.console.print(f"\n[bold]{header}:[/bold]")

        # Prompt snippet in a Rich Panel
        if sp.render_error:
            self.console.print(
                Panel(
                    f"[red]{sp.render_error}[/red]",
                    title="Prompt",
                    border_style="dim",
                    expand=False,
                )
            )
        else:
            self.console.print("  Prompt:")
            self.console.print(
                Panel(
                    sp.prompt_snippet,
                    border_style="dim",
                    expand=False,
                )
            )

        # Validations
        if sp.expanded_validations:
            self.console.print(
                f"  Validations (expanded for sheet {sp.sheet_num}):"
            )
            for ev in sp.expanded_validations:
                num = ev.index + 1
                path_display = ev.expanded_path or ""
                if not ev.applicable:
                    self.console.print(
                        f"    [dim]{num}. {ev.type}: {path_display}"
                        f" (not applicable: {ev.condition})[/dim]"
                    )
                else:
                    self.console.print(
                        f"    {num}. {ev.type}: {path_display}"
                    )

    # Render errors summary
    if preview.render_errors:
        self.console.print("\n[red bold]Render Errors:[/red bold]")
        for err in preview.render_errors:
            self.console.print(f"  [red]{err}[/red]")

report_rendering_json

report_rendering_json(preview)

Return a rendering preview as a JSON-serializable dict.

Parameters:

Name	Type	Description	Default
preview	RenderingPreview	The rendering preview to serialize.	required

Returns:

Type	Description
dict[str, Any]	Dict with total_sheets, has_fan_out, has_dependencies,
dict[str, Any]	render_errors, and a sheets array.

Source code in src/marianne/validation/reporter.py

def report_rendering_json(self, preview: "RenderingPreview") -> dict[str, Any]:
    """Return a rendering preview as a JSON-serializable dict.

    Args:
        preview: The rendering preview to serialize.

    Returns:
        Dict with total_sheets, has_fan_out, has_dependencies,
        render_errors, and a sheets array.
    """
    sheets: list[dict[str, Any]] = []
    for sp in preview.sheets:
        validations: list[dict[str, Any]] = []
        for ev in sp.expanded_validations:
            v: dict[str, Any] = {
                "index": ev.index,
                "type": ev.type,
                "applicable": ev.applicable,
            }
            if ev.description:
                v["description"] = ev.description
            if ev.expanded_path:
                v["expanded_path"] = ev.expanded_path
            if ev.pattern:
                v["pattern"] = ev.pattern
            if ev.condition:
                v["condition"] = ev.condition
            validations.append(v)

        sheet_dict: dict[str, Any] = {
            SHEET_NUM_KEY: sp.sheet_num,
            "item_range": list(sp.item_range),
            "stage": sp.stage,
            "instance": sp.instance,
            "fan_count": sp.fan_count,
            "prompt_snippet": sp.prompt_snippet,
            "render_error": sp.render_error,
            "validations": validations,
        }
        sheets.append(sheet_dict)

    return {
        "total_sheets": preview.total_sheets,
        "has_fan_out": preview.has_fan_out,
        "has_dependencies": preview.has_dependencies,
        "render_errors": preview.render_errors,
        "sheets": sheets,
    }

format_plain

format_plain(issues)

Format issues as plain text for logs.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue]	List of validation issues	required

Returns:

Type	Description
str	Plain text representation

Source code in src/marianne/validation/reporter.py

def format_plain(self, issues: list[ValidationIssue]) -> str:
    """Format issues as plain text for logs.

    Args:
        issues: List of validation issues

    Returns:
        Plain text representation
    """
    if not issues:
        return "Validation passed: no issues found"

    lines = []
    for issue in issues:
        severity = issue.severity.value.upper()
        line_info = f" (line {issue.line})" if issue.line else ""
        lines.append(f"[{severity}] {issue.check_id}{line_info}: {issue.message}")

        if issue.suggestion:
            lines.append(f"    Suggestion: {issue.suggestion}")

    # Summary
    errors = sum(1 for i in issues if i.severity == ValidationSeverity.ERROR)
    warnings = sum(1 for i in issues if i.severity == ValidationSeverity.WARNING)
    infos = sum(1 for i in issues if i.severity == ValidationSeverity.INFO)

    lines.append("")
    lines.append(f"Total: {errors} errors, {warnings} warnings, {infos} info")
    lines.append(f"Validation: {'FAILED' if errors else 'PASSED'}")

    return "\n".join(lines)

ValidationRunner

ValidationRunner(checks=None)

Orchestrates validation checks against a configuration.

The runner collects issues from all checks, sorts them by severity, and provides summary information for reporting.

Initialize with a list of checks.

Parameters:

Name	Type	Description	Default
checks	list[ValidationCheck] | None	List of validation checks to run. If None, uses default checks.	None

Source code in src/marianne/validation/runner.py

def __init__(self, checks: list[ValidationCheck] | None = None):
    """Initialize with a list of checks.

    Args:
        checks: List of validation checks to run. If None, uses default checks.
    """
    self._checks: list[ValidationCheck] = checks or []

Functions

add_check

add_check(check)

Add a check to the runner.

Source code in src/marianne/validation/runner.py

def add_check(self, check: ValidationCheck) -> None:
    """Add a check to the runner."""
    self._checks.append(check)

validate

validate(config, config_path, raw_yaml)

Run all validation checks against the configuration.

Parameters:

Name	Type	Description	Default
config	JobConfig	Parsed JobConfig object	required
config_path	Path	Path to the config file	required
raw_yaml	str	Raw YAML text for line number extraction	required

Returns:

Type	Description
list[ValidationIssue]	List of all issues found, sorted by severity (errors first)

Source code in src/marianne/validation/runner.py

def validate(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Run all validation checks against the configuration.

    Args:
        config: Parsed JobConfig object
        config_path: Path to the config file
        raw_yaml: Raw YAML text for line number extraction

    Returns:
        List of all issues found, sorted by severity (errors first)
    """
    all_issues: list[ValidationIssue] = []

    for check in self._checks:
        try:
            issues = check.check(config, config_path, raw_yaml)
            all_issues.extend(issues)
        except Exception as e:
            # Don't let one check failure break all validation
            all_issues.append(
                ValidationIssue(
                    check_id=check.check_id,
                    severity=ValidationSeverity.ERROR,
                    message=f"Check {check.check_id} failed to execute: {e}",
                    suggestion="This may be a bug in Marianne validation",
                )
            )

    # Sort by severity (ERROR > WARNING > INFO)
    severity_order = {
        ValidationSeverity.ERROR: 0,
        ValidationSeverity.WARNING: 1,
        ValidationSeverity.INFO: 2,
    }
    all_issues.sort(key=lambda i: severity_order.get(i.severity, 3))

    return all_issues

get_exit_code

get_exit_code(issues)

Determine exit code based on issues found.

Returns:

Name	Type	Description
0	int	No errors (warnings/info OK)
1	int	One or more ERROR-severity issues

Source code in src/marianne/validation/runner.py

def get_exit_code(self, issues: list[ValidationIssue]) -> int:
    """Determine exit code based on issues found.

    Returns:
        0: No errors (warnings/info OK)
        1: One or more ERROR-severity issues
    """
    return 1 if self.has_errors(issues) else 0

has_errors

has_errors(issues)

Check if any issues are errors.

Source code in src/marianne/validation/runner.py

def has_errors(self, issues: list[ValidationIssue]) -> bool:
    """Check if any issues are errors."""
    return any(i.severity == ValidationSeverity.ERROR for i in issues)

count_by_severity

count_by_severity(issues)

Count issues by severity level.

Source code in src/marianne/validation/runner.py

def count_by_severity(
    self, issues: list[ValidationIssue]
) -> dict[ValidationSeverity, int]:
    """Count issues by severity level."""
    counts: dict[ValidationSeverity, int] = {
        ValidationSeverity.ERROR: 0,
        ValidationSeverity.WARNING: 0,
        ValidationSeverity.INFO: 0,
    }
    for issue in issues:
        counts[issue.severity] += 1
    return counts

Functions

create_default_checks

create_default_checks()

Create the default set of validation checks.

Returns all built-in checks in recommended execution order.

Source code in src/marianne/validation/runner.py

def create_default_checks() -> list[ValidationCheck]:
    """Create the default set of validation checks.

    Returns all built-in checks in recommended execution order.
    """
    # Note: We return instances, not classes
    # The type checker sees these as ValidationCheck protocol implementations
    checks: list[ValidationCheck] = [
        # Jinja checks (most common issues)
        JinjaSyntaxCheck(),
        JinjaUndefinedVariableCheck(),
        # Path checks
        WorkspaceParentExistsCheck(),
        TemplateFileExistsCheck(),
        SystemPromptFileCheck(),
        WorkingDirectoryCheck(),
        SkillFilesExistCheck(),
        PreludeCadenzaFileCheck(),
        # Config checks
        RegexPatternCheck(),
        ValidationTypeCheck(),
        TimeoutRangeCheck(),
        EmptyPatternCheck(),
        VersionReferenceCheck(),
        # Instrument checks
        InstrumentNameCheck(),
        InstrumentFallbackCheck(),
        # Best-practice checks
        JinjaInValidationPathCheck(),
        FormatSyntaxInTemplateCheck(),
        NoValidationsCheck(),
        MissingSkipPermissionsCheck(),
        FileExistsOnlyCheck(),
        FanOutWithoutDependenciesCheck(),
        FanOutWithoutParallelCheck(),
        VariableShadowingCheck(),
        MissingDisableMcpCheck(),
        SkipWhenSheetRangeCheck(),
    ]
    return checks