engine

engine

Validation engine — executes validation rules against sheet outputs.

Dispatches to type-specific check methods: file_exists, file_modified, content_contains, content_regex, and command_succeeds.

Attributes

Classes

ValidationEngine

ValidationEngine(workspace, sheet_context)

Executes validation rules against sheet outputs.

Handles path template expansion and dispatches to type-specific validation methods.

Initialize validation engine.

Source code in src/marianne/execution/validation/engine.py

def __init__(self, workspace: Path, sheet_context: dict[str, Any]) -> None:
    """Initialize validation engine."""
    self.workspace = workspace.resolve()
    self.sheet_context = sheet_context
    self._mtime_tracker = FileModificationTracker()

Functions

expand_path

expand_path(path_template)

Expand path template with sheet context variables.

Supports: {sheet_num}, {workspace}, {start_item}, {end_item}

Both workspace-relative and absolute paths are allowed. Agents work in backend.working_directory (typically the project root) and create files there — restricting validations to the workspace directory would prevent checking those files.

Source code in src/marianne/execution/validation/engine.py

def expand_path(self, path_template: str) -> Path:
    """Expand path template with sheet context variables.

    Supports: {sheet_num}, {workspace}, {start_item}, {end_item}

    Both workspace-relative and absolute paths are allowed. Agents work
    in ``backend.working_directory`` (typically the project root) and
    create files there — restricting validations to the workspace
    directory would prevent checking those files.
    """
    context = dict(self.sheet_context)
    context["workspace"] = str(self.workspace)

    try:
        expanded = path_template.format(**context)
    except IndexError as exc:
        raise ValueError(
            f"Invalid path template '{path_template}': {exc}. "
            "Use named placeholders like {{workspace}}, not bare {{}}."
        ) from exc
    return Path(expanded).resolve()

snapshot_mtime_files

snapshot_mtime_files(rules)

Snapshot mtimes for all file_modified rules before sheet execution.

Source code in src/marianne/execution/validation/engine.py

def snapshot_mtime_files(self, rules: list[ValidationRule]) -> None:
    """Snapshot mtimes for all file_modified rules before sheet execution."""
    paths = [
        self.expand_path(r.path)
        for r in rules
        if r.type == "file_modified" and r.path
    ]
    self._mtime_tracker.snapshot(paths)

get_applicable_rules

get_applicable_rules(rules)

Get rules that apply to the current sheet context.

Source code in src/marianne/execution/validation/engine.py

def get_applicable_rules(
    self, rules: list[ValidationRule]
) -> list[ValidationRule]:
    """Get rules that apply to the current sheet context."""
    return [r for r in rules if self._check_condition(r.condition)]

run_validations async

run_validations(rules)

Execute all validation rules and return aggregate result.

Source code in src/marianne/execution/validation/engine.py

async def run_validations(self, rules: list[ValidationRule]) -> SheetValidationResult:
    """Execute all validation rules and return aggregate result."""
    applicable_rules = self.get_applicable_rules(rules)
    results: list[ValidationResult] = []

    for rule in applicable_rules:
        result = await self._run_single_validation(rule)
        results.append(result)

    return SheetValidationResult(
        sheet_num=self.sheet_context.get(SHEET_NUM_KEY, 0),
        results=results,
        rules_checked=len(applicable_rules),
    )

run_staged_validations async

run_staged_validations(rules)

Execute validations in stage order with fail-fast behavior.

Source code in src/marianne/execution/validation/engine.py

async def run_staged_validations(
    self, rules: list[ValidationRule]
) -> tuple[SheetValidationResult, int | None]:
    """Execute validations in stage order with fail-fast behavior."""
    applicable_rules = self.get_applicable_rules(rules)

    if not applicable_rules:
        return SheetValidationResult(
            sheet_num=self.sheet_context.get(SHEET_NUM_KEY, 0),
            results=[],
            rules_checked=0,
        ), None

    stages: dict[int, list[ValidationRule]] = defaultdict(list)
    for rule in applicable_rules:
        stages[rule.stage].append(rule)

    all_results: list[ValidationResult] = []
    failed_stage: int | None = None

    for stage_num in sorted(stages.keys()):
        stage_rules = stages[stage_num]
        stage_passed = True

        for rule in stage_rules:
            result = await self._run_single_validation(rule)
            all_results.append(result)
            if not result.passed:
                stage_passed = False

        if not stage_passed:
            failed_stage = stage_num
            self._mark_remaining_stages_skipped(
                stages, stage_num, all_results
            )
            break

    return SheetValidationResult(
        sheet_num=self.sheet_context.get(SHEET_NUM_KEY, 0),
        results=all_results,
        rules_checked=len(applicable_rules),
    ), failed_stage