Skip to content

Index

validation

Validation framework for sheet outputs.

Re-exports all public names from the subpackage modules so that existing imports like from marianne.execution.validation import X continue to work after the monolith split.

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 
43
44
45
46
47
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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
102
103
104
105
106
107
108
109
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 
155
156
157
158
159
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 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
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

FailureHistoryStore 

FailureHistoryStore(state)

Queries past validation failures from checkpoint state.

Enables history-aware prompt generation by extracting validation failures from previous sheets and finding similar failures.

Initialize failure history store.

Source code in src/marianne/execution/validation/history.py 
35
36
37
def __init__(self, state: CheckpointState) -> None:
    """Initialize failure history store."""
    self._state = state
Functions
query_similar_failures 
query_similar_failures(current_sheet, rule_types=None, failure_categories=None, limit=3)

Query past validation failures similar to expected patterns.

Source code in src/marianne/execution/validation/history.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def query_similar_failures(
    self,
    current_sheet: int,
    rule_types: list[str] | None = None,
    failure_categories: list[str] | None = None,
    limit: int = 3,
) -> list[HistoricalFailure]:
    """Query past validation failures similar to expected patterns."""
    failures: list[HistoricalFailure] = []

    for sheet_num in sorted(self._state.sheets.keys(), reverse=True):
        if sheet_num >= current_sheet:
            continue

        sheet = self._state.sheets.get(sheet_num)
        if not sheet or not sheet.validation_details:
            continue

        for detail in sheet.validation_details:
            if detail.get("passed", False):
                continue

            rule_type = detail.get("rule_type", "")
            failure_category = detail.get("failure_category")

            if rule_types and rule_type not in rule_types:
                continue
            if failure_categories and failure_category not in failure_categories:
                continue

            failures.append(self._detail_to_failure(sheet_num, detail))

            if len(failures) >= limit:
                return failures

    return failures
query_recent_failures 
query_recent_failures(current_sheet, lookback_sheets=3, limit=3)

Query recent validation failures from nearby sheets.

Source code in src/marianne/execution/validation/history.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def query_recent_failures(
    self,
    current_sheet: int,
    lookback_sheets: int = 3,
    limit: int = 3,
) -> list[HistoricalFailure]:
    """Query recent validation failures from nearby sheets."""
    failures: list[HistoricalFailure] = []

    for offset in range(1, lookback_sheets + 1):
        sheet_num = current_sheet - offset
        if sheet_num <= 0:
            break

        sheet = self._state.sheets.get(sheet_num)
        if not sheet or not sheet.validation_details:
            continue

        for detail in sheet.validation_details:
            if detail.get("passed", False):
                continue

            failures.append(self._detail_to_failure(sheet_num, detail))

            if len(failures) >= limit:
                return failures

    return failures
has_failures 
has_failures(current_sheet)

Check if there are any historical failures to query.

Source code in src/marianne/execution/validation/history.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
def has_failures(self, current_sheet: int) -> bool:
    """Check if there are any historical failures to query."""
    for sheet_num, sheet in self._state.sheets.items():
        if sheet_num >= current_sheet:
            continue

        if not sheet.validation_details:
            continue

        if any(not d.get("passed", False) for d in sheet.validation_details):
            return True

    return False

HistoricalFailure dataclass 

HistoricalFailure(sheet_num, rule_type, description, failure_reason=None, failure_category=None, suggested_fix=None)

A single historical validation failure for prompt injection.

FileModificationTracker 

FileModificationTracker()

Tracks file mtimes before sheet execution for file_modified checks.

Source code in src/marianne/execution/validation/models.py 
223
224
def __init__(self) -> None:
    self._mtimes: dict[str, float] = {}
Functions
snapshot 
snapshot(paths)

Capture mtimes of files before sheet execution.

Source code in src/marianne/execution/validation/models.py 
226
227
228
229
230
231
232
233
def snapshot(self, paths: list[Path]) -> None:
    """Capture mtimes of files before sheet execution."""
    for path in paths:
        path_str = str(path.resolve())
        if path.exists():
            self._mtimes[path_str] = path.stat().st_mtime
        else:
            self._mtimes[path_str] = 0.0
was_modified 
was_modified(path)

Check if file was modified (or created) after snapshot.

Source code in src/marianne/execution/validation/models.py 
235
236
237
238
239
240
241
242
243
def was_modified(self, path: Path) -> bool:
    """Check if file was modified (or created) after snapshot."""
    resolved = path.resolve()
    try:
        current_mtime = resolved.stat().st_mtime
    except (OSError, ValueError):
        return False
    original_mtime = self._mtimes.get(str(resolved), 0.0)
    return current_mtime > original_mtime
get_original_mtime 
get_original_mtime(path)

Get the original mtime from snapshot.

Source code in src/marianne/execution/validation/models.py 
245
246
247
248
def get_original_mtime(self, path: Path) -> float | None:
    """Get the original mtime from snapshot."""
    path_str = str(path.resolve())
    return self._mtimes.get(path_str)
clear 
clear()

Clear all tracked mtimes.

Source code in src/marianne/execution/validation/models.py 
250
251
252
def clear(self) -> None:
    """Clear all tracked mtimes."""
    self._mtimes.clear()

SheetValidationResult dataclass 

SheetValidationResult(sheet_num, results, rules_checked=0)

Aggregate result of all validations for a sheet.

Attributes
all_passed property 
all_passed

Check if all validations passed.

passed_count property 
passed_count

Count of passed validations.

failed_count property 
failed_count

Count of failed validations (excluding skipped).

skipped_count property 
skipped_count

Count of skipped validations (due to staged fail-fast).

executed_count property 
executed_count

Count of validations that actually executed (not skipped).

pass_percentage property 
pass_percentage

Percentage of validations that passed.

executed_pass_percentage property 
executed_pass_percentage

Percentage of EXECUTED validations that passed.

majority_passed property 
majority_passed

Returns True if >50% of validations passed.

aggregate_confidence property 
aggregate_confidence

Calculate weighted aggregate confidence across all validation results.

Functions
get_passed_rules 
get_passed_rules()

Get rules that passed.

Source code in src/marianne/execution/validation/models.py 
157
158
159
def get_passed_rules(self) -> list[ValidationRule]:
    """Get rules that passed."""
    return [result.rule for result in self.results if result.passed]
get_failed_rules 
get_failed_rules()

Get rules that failed.

Source code in src/marianne/execution/validation/models.py 
161
162
163
def get_failed_rules(self) -> list[ValidationRule]:
    """Get rules that failed."""
    return [result.rule for result in self.results if not result.passed]
get_passed_results 
get_passed_results()

Get results that passed.

Source code in src/marianne/execution/validation/models.py 
165
166
167
def get_passed_results(self) -> list[ValidationResult]:
    """Get results that passed."""
    return [result for result in self.results if result.passed]
get_failed_results 
get_failed_results()

Get results that failed.

Source code in src/marianne/execution/validation/models.py 
169
170
171
def get_failed_results(self) -> list[ValidationResult]:
    """Get results that failed."""
    return [result for result in self.results if not result.passed]
to_dict_list 
to_dict_list()

Convert all results to serializable list.

Source code in src/marianne/execution/validation/models.py 
173
174
175
def to_dict_list(self) -> list[ValidationDetailDict]:
    """Convert all results to serializable list."""
    return [result.to_dict() for result in self.results]
get_semantic_summary 
get_semantic_summary()

Aggregate semantic information from failed validations.

Source code in src/marianne/execution/validation/models.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def get_semantic_summary(self) -> dict[str, Any]:
    """Aggregate semantic information from failed validations."""
    category_counts: dict[str, int] = {}
    has_semantic_info = False

    for result in self.results:
        if not result.passed and result.failure_category:
            has_semantic_info = True
            category = result.failure_category
            category_counts[category] = category_counts.get(category, 0) + 1

    dominant_category: str | None = None
    if category_counts:
        dominant_category = max(category_counts, key=lambda k: category_counts[k])

    return {
        "category_counts": category_counts,
        "dominant_category": dominant_category,
        "has_semantic_info": has_semantic_info,
        "total_failures": self.failed_count,
    }
get_actionable_hints 
get_actionable_hints(limit=3)

Extract actionable hints from failed validations.

Source code in src/marianne/execution/validation/models.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def get_actionable_hints(self, limit: int = 3) -> list[str]:
    """Extract actionable hints from failed validations."""
    hints: list[str] = []
    seen: set[str] = set()

    for result in self.results:
        if not result.passed and result.suggested_fix:
            hint = result.suggested_fix
            if len(hint) > 100:
                hint = hint[:97] + "..."

            if hint not in seen:
                seen.add(hint)
                hints.append(hint)

            if len(hints) >= limit:
                break

    return hints

ValidationResult dataclass 

ValidationResult(rule, passed, actual_value=None, expected_value=None, error_message=None, checked_at=utc_now(), check_duration_ms=0.0, confidence=1.0, confidence_factors=dict(), failure_reason=None, failure_category=None, suggested_fix=None, error_type=None)

Result of a single validation check.

Attributes
confidence class-attribute instance-attribute 
confidence = 1.0

Confidence in this validation result (0.0-1.0). Default 1.0 = fully confident.

confidence_factors class-attribute instance-attribute 
confidence_factors = field(default_factory=dict)

Factors affecting confidence, e.g., {'file_age': 0.9, 'pattern_specificity': 0.8}.

failure_reason class-attribute instance-attribute 
failure_reason = None

Semantic explanation of why validation failed.

failure_category class-attribute instance-attribute 
failure_category = None

Category of failure: 'missing', 'malformed', 'incomplete', 'stale', 'error'.

suggested_fix class-attribute instance-attribute 
suggested_fix = None

Hint for how to fix the issue.

error_type class-attribute instance-attribute 
error_type = None

Distinguishes validation failures from validation crashes. None or 'validation_failure' = output didn't meet the rule. 'internal_error' = the validation check itself crashed.

Functions
to_dict 
to_dict()

Convert to serializable dictionary.

Source code in src/marianne/execution/validation/models.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def to_dict(self) -> ValidationDetailDict:
    """Convert to serializable dictionary."""
    return {
        "rule_type": self.rule.type,
        "description": self.rule.description,
        "path": self.rule.path,
        "pattern": self.rule.pattern,
        "passed": self.passed,
        "actual_value": self.actual_value,
        "expected_value": self.expected_value,
        "error_message": self.error_message,
        "checked_at": self.checked_at.isoformat(),
        "check_duration_ms": self.check_duration_ms,
        "confidence": self.confidence,
        "confidence_factors": self.confidence_factors,
        "failure_reason": self.failure_reason,
        "failure_category": self.failure_category,
        "suggested_fix": self.suggested_fix,
        "error_type": self.error_type,
    }
format_failure_summary 
format_failure_summary()

Format failure information for prompt injection.

Source code in src/marianne/execution/validation/models.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def format_failure_summary(self) -> str:
    """Format failure information for prompt injection."""
    if self.passed:
        return ""

    parts: list[str] = []
    if self.failure_category:
        parts.append(f"[{self.failure_category.upper()}]")
    if self.failure_reason:
        parts.append(self.failure_reason)
    if self.suggested_fix:
        parts.append(f"Fix: {self.suggested_fix}")

    return " ".join(parts)

KeyVariable dataclass 

KeyVariable(key, value, source_line='', line_number=0)

A key-value pair extracted from sheet output.

KeyVariableExtractor 

KeyVariableExtractor(key_filter=None, case_sensitive=False)

Extracts key-value pairs from sheet output content.

Initialize extractor.

Source code in src/marianne/execution/validation/semantic.py 
103
104
105
106
107
108
109
110
def __init__(
    self,
    key_filter: list[str] | None = None,
    case_sensitive: bool = False,
) -> None:
    """Initialize extractor."""
    self.key_filter = key_filter
    self.case_sensitive = case_sensitive
Functions
extract 
extract(content)

Extract key-value pairs from content.

Source code in src/marianne/execution/validation/semantic.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def extract(self, content: str) -> list[KeyVariable]:
    """Extract key-value pairs from content."""
    if not content:
        return []

    variables: list[KeyVariable] = []
    seen_keys: set[str] = set()

    lines = content.split("\n")
    line_map: dict[str, int] = {}
    for i, line in enumerate(lines, 1):
        line_map[line] = i

    for match in self._KEY_VALUE_PATTERN.finditer(content):
        key = match.group(1)
        value = match.group(2).strip()
        source_line = match.group(0)

        if self._should_include(key) and key not in seen_keys:
            variables.append(KeyVariable(
                key=key,
                value=value,
                source_line=source_line,
                line_number=line_map.get(source_line, 0),
            ))
            seen_keys.add(key)

    return variables

SemanticConsistencyChecker 

SemanticConsistencyChecker(extractor=None, strict_mode=False)

Checks semantic consistency between sequential sheet outputs.

Initialize checker.

Source code in src/marianne/execution/validation/semantic.py 
156
157
158
159
160
161
162
163
def __init__(
    self,
    extractor: KeyVariableExtractor | None = None,
    strict_mode: bool = False,
) -> None:
    """Initialize checker."""
    self.extractor = extractor or KeyVariableExtractor()
    self.strict_mode = strict_mode
Functions
check_consistency 
check_consistency(sheet_outputs, sequential_only=True)

Check semantic consistency across sheet outputs.

Source code in src/marianne/execution/validation/semantic.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def check_consistency(
    self,
    sheet_outputs: dict[int, str],
    sequential_only: bool = True,
) -> SemanticConsistencyResult:
    """Check semantic consistency across sheet outputs."""
    result = SemanticConsistencyResult(
        sheets_compared=sorted(sheet_outputs.keys()),
    )

    if len(sheet_outputs) < 2:
        return result

    sheet_variables: dict[int, dict[str, KeyVariable]] = {}
    for sheet_num, content in sheet_outputs.items():
        variables = self.extractor.extract(content)
        sheet_variables[sheet_num] = {v.key: v for v in variables}

    all_keys: set[str] = set()
    for vars_dict in sheet_variables.values():
        all_keys.update(vars_dict.keys())
    result.keys_checked = len(all_keys)

    sheets_sorted = sorted(sheet_outputs.keys())

    if sequential_only:
        for i in range(len(sheets_sorted) - 1):
            sheet_a = sheets_sorted[i]
            sheet_b = sheets_sorted[i + 1]
            self._compare_sheets(
                sheet_a, sheet_variables[sheet_a],
                sheet_b, sheet_variables[sheet_b],
                result,
            )
    else:
        for key in all_keys:
            value_groups: dict[str, list[int]] = defaultdict(list)
            for sheet_num in sheets_sorted:
                var = sheet_variables[sheet_num].get(key)
                if var is not None:
                    value_groups[var.value.lower()].append(sheet_num)

            if len(value_groups) <= 1:
                continue

            group_list = list(value_groups.values())
            for gi in range(len(group_list)):
                for gj in range(gi + 1, len(group_list)):
                    for sheet_a in group_list[gi]:
                        for sheet_b in group_list[gj]:
                            var_a = sheet_variables[sheet_a][key]
                            var_b = sheet_variables[sheet_b][key]
                            result.inconsistencies.append(SemanticInconsistency(
                                key=key,
                                sheet_a=sheet_a,
                                value_a=var_a.value,
                                sheet_b=sheet_b,
                                value_b=var_b.value,
                                severity="error" if self.strict_mode else "warning",
                            ))

    return result

SemanticConsistencyResult dataclass 

SemanticConsistencyResult(sheets_compared=list(), inconsistencies=list(), keys_checked=0, checked_at=utc_now())

Result of cross-sheet semantic consistency check.

Attributes
is_consistent property 
is_consistent

True if no inconsistencies were found.

error_count property 
error_count

Count of error-severity inconsistencies.

warning_count property 
warning_count

Count of warning-severity inconsistencies.

Functions
to_dict 
to_dict()

Convert to serializable dictionary.

Source code in src/marianne/execution/validation/semantic.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_dict(self) -> dict[str, Any]:
    """Convert to serializable dictionary."""
    return {
        "sheets_compared": self.sheets_compared,
        "inconsistencies": [
            {
                "key": i.key,
                "sheet_a": i.sheet_a,
                "value_a": i.value_a,
                "sheet_b": i.sheet_b,
                "value_b": i.value_b,
                "severity": i.severity,
            }
            for i in self.inconsistencies
        ],
        "keys_checked": self.keys_checked,
        "checked_at": self.checked_at.isoformat(),
        "is_consistent": self.is_consistent,
        "error_count": self.error_count,
        "warning_count": self.warning_count,
    }

SemanticInconsistency dataclass 

SemanticInconsistency(key, sheet_a, value_a, sheet_b, value_b, severity='warning')

Represents a semantic inconsistency between sheets.

Functions
format_message 
format_message()

Format as human-readable message.

Source code in src/marianne/execution/validation/semantic.py 
39
40
41
42
43
44
45
def format_message(self) -> str:
    """Format as human-readable message."""
    return (
        f"Key '{self.key}' has inconsistent values: "
        f"sheet {self.sheet_a}='{self.value_a}' vs "
        f"sheet {self.sheet_b}='{self.value_b}'"
    )