config

config

Configuration structure validation checks.

Validates configuration values like regex patterns, timeout ranges, validation rule completeness, and instrument name resolution.

Classes

RegexPatternCheck

Check that regex patterns in validations compile (V007).

Invalid regex patterns will cause runtime errors during validation.

Functions

check

check(config, config_path, raw_yaml)

Check regex patterns in validations and rate limit detection.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check regex patterns in validations and rate limit detection."""
    issues: list[ValidationIssue] = []

    # Check validation rule patterns
    for i, validation in enumerate(config.validations):
        if validation.type == "content_regex" and validation.pattern:
            issue = self._check_pattern(
                validation.pattern,
                f"validation[{i}].pattern",
                self._find_pattern_line(raw_yaml, validation.pattern),
            )
            if issue:
                issues.append(issue)

    # Check rate limit detection patterns
    for i, pattern in enumerate(config.rate_limit.detection_patterns):
        issue = self._check_pattern(
            pattern,
            f"rate_limit.detection_patterns[{i}]",
            None,
        )
        if issue:
            issues.append(issue)

    return issues

ValidationTypeCheck

Check that validation rules have required fields (V008).

Ensures validations have the fields needed for their type.

Functions

check

check(config, config_path, raw_yaml)

Check validation rules have required fields.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check validation rules have required fields."""
    issues: list[ValidationIssue] = []

    required_fields = {
        "file_exists": ["path"],
        "file_modified": ["path"],
        "content_contains": ["path", "pattern"],
        "content_regex": ["path", "pattern"],
        "command_succeeds": ["command"],
    }

    for i, validation in enumerate(config.validations):
        required = required_fields.get(validation.type, [])
        missing = []

        for field in required:
            value = getattr(validation, field, None)
            if value is None or (isinstance(value, str) and not value.strip()):
                missing.append(field)

        if missing:
            issues.append(
                ValidationIssue(
                    check_id=self.check_id,
                    severity=self.severity,
                    message=(
                        f"Validation rule {i + 1} ({validation.type})"
                        f" missing required fields: {', '.join(missing)}"
                    ),
                    suggestion=f"Add {', '.join(missing)} to the validation rule",
                    metadata={
                        "validation_index": str(i),
                        "validation_type": validation.type,
                        "missing_fields": ",".join(missing),
                    },
                )
            )

    return issues

TimeoutRangeCheck

Check timeout values are reasonable (V103/V104).

Warns about very short timeouts (may cause failures) or very long timeouts (may waste resources).

Functions

check

check(config, config_path, raw_yaml)

Check timeout values.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check timeout values."""
    issues: list[ValidationIssue] = []

    timeout = config.backend.timeout_seconds

    if timeout < self.MIN_REASONABLE_TIMEOUT:
        issues.append(
            ValidationIssue(
                check_id="V103",
                severity=ValidationSeverity.WARNING,
                message=f"Very short timeout ({timeout}s) may cause premature failures",
                line=find_line_in_yaml(raw_yaml, "timeout_seconds:"),
                suggestion=(
                    f"Consider timeout_seconds >="
                    f" {self.MIN_REASONABLE_TIMEOUT}"
                    f" for Claude CLI operations"
                ),
                metadata={
                    "timeout": str(timeout),
                    "threshold": str(self.MIN_REASONABLE_TIMEOUT),
                },
            )
        )

    if timeout > self.MAX_REASONABLE_TIMEOUT:
        issues.append(
            ValidationIssue(
                check_id="V104",
                severity=ValidationSeverity.INFO,
                message=(
                    f"Very long timeout ({timeout}s = {timeout / 3600:.1f}h)"
                    f" - consider if this is necessary"
                ),
                line=find_line_in_yaml(raw_yaml, "timeout_seconds:"),
                suggestion=(
                    "Long timeouts can tie up resources;"
                    " consider breaking into smaller tasks"
                ),
                metadata={
                    "timeout": str(timeout),
                    "threshold": str(self.MAX_REASONABLE_TIMEOUT),
                },
            )
        )

    return issues

VersionReferenceCheck

Check that evolved scores don't reference previous version paths (V009).

When a score evolves (e.g., v20 → v21), the new score file should have all references updated to the new version. This catches cases where the name, workspace, or other paths still reference the old version.

Functions

check

check(config, config_path, raw_yaml)

Check for stale version references in evolved scores.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check for stale version references in evolved scores."""
    issues: list[ValidationIssue] = []

    # Extract version from filename (e.g., v21 from marianne-opus-evolution-v21.yaml)
    filename = config_path.name
    version_match = re.search(r"-v(\d+)\.yaml$", filename)
    if not version_match:
        # Not a versioned score file, skip this check
        return issues

    current_version = int(version_match.group(1))
    if current_version <= 1:
        # v1 has no previous version to check against
        return issues

    previous_version = current_version - 1
    prev_patterns = [
        f"-v{previous_version}",  # e.g., -v20
        f"v{previous_version}.0",  # e.g., v20.0
        f"v{previous_version}/",  # e.g., workspace-v20/
        f"-v{previous_version}/",  # e.g., evolution-workspace-v20/
        f"evolution-v{previous_version}",  # e.g., evolution-v20
    ]

    # Check the job name
    if f"-v{previous_version}" in config.name:
        issues.append(
            ValidationIssue(
                check_id=self.check_id,
                severity=self.severity,
                message=(
                    f"Job name '{config.name}' references"
                    f" v{previous_version} but file is v{current_version}"
                ),
                line=find_line_in_yaml(raw_yaml, "name:"),
                suggestion=f"Update name to use v{current_version}",
                metadata={
                    "field": "name",
                    "current_version": str(current_version),
                    "previous_version": str(previous_version),
                },
            )
        )

    # Check the workspace path
    workspace_str = str(config.workspace)
    if f"-v{previous_version}" in workspace_str or f"v{previous_version}/" in workspace_str:
        issues.append(
            ValidationIssue(
                check_id=self.check_id,
                severity=self.severity,
                message=(
                    f"Workspace path references v{previous_version}"
                    f" but file is v{current_version}"
                ),
                line=find_line_in_yaml(raw_yaml, "workspace:"),
                context=workspace_str,
                suggestion=f"Update workspace to use v{current_version}",
                metadata={
                    "field": "workspace",
                    "current_version": str(current_version),
                    "previous_version": str(previous_version),
                },
            )
        )

    # Scan raw YAML for other references (in comments, descriptions, etc.)
    for i, line in enumerate(raw_yaml.split("\n"), 1):
        # Skip lines that are defining the version progression history
        if "VERSION PROGRESSION:" in line or "→" in line or "->" in line:
            continue
        # Skip lines in comments that are documenting history
        if line.strip().startswith("#") and ("v" + str(previous_version - 1) in line):
            continue

        for pattern in prev_patterns:
            if pattern in line:
                # Check if this is just historical documentation
                hist_markers = [
                    "EVOLUTION FROM", "evolved from", "LEARNINGS",
                ]
                if any(hist in line for hist in hist_markers):
                    continue
                # Check if it's in the version progression list
                if (
                    f"v{previous_version}→v{current_version}" in line
                    or f"v{previous_version}->v{current_version}" in line
                ):
                    continue

                issues.append(
                    ValidationIssue(
                        check_id=self.check_id,
                        severity=ValidationSeverity.WARNING,
                        message=(
                            f"Line {i} references v{previous_version}"
                            f" - verify this is intentional"
                        ),
                        line=i,
                        context=line.strip()[:80],
                        suggestion=(
                    f"Update to v{current_version} if this should"
                    f" reference the current version"
                ),
                        metadata={
                            "pattern": pattern,
                            "current_version": str(current_version),
                            "previous_version": str(previous_version),
                        },
                    )
                )
                break  # Only one issue per line

    return issues

EmptyPatternCheck

Check for empty patterns in validations (V106).

Functions

check

check(config, config_path, raw_yaml)

Check for empty patterns.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check for empty patterns."""
    issues: list[ValidationIssue] = []

    for i, validation in enumerate(config.validations):
        if (
            validation.type in ("content_contains", "content_regex")
            and validation.pattern is not None
            and validation.pattern.strip() == ""
        ):
            issues.append(
                ValidationIssue(
                    check_id=self.check_id,
                    severity=self.severity,
                    message=(
                        f"Empty pattern in validation rule {i + 1}"
                        f" will match any content"
                    ),
                    suggestion="Add a meaningful pattern or remove this validation",
                    metadata={
                        "validation_index": str(i),
                    },
                )
            )

    return issues

InstrumentNameCheck

Check that instrument names resolve to known profiles (V210).

Warns when an instrument name doesn't match any loaded instrument profile. This catches typos (e.g., 'clause-code' instead of 'claude-code') at validation time rather than runtime.

Severity is WARNING (not ERROR) because the conductor may have instruments the validator doesn't know about — profiles loaded from other directories, dynamic instruments, etc. The warning is informational, not blocking.

Checks: - Top-level instrument: field - sheet.per_sheet_instruments per-sheet overrides - sheet.instrument_map batch assignments - movements.N.instrument per-movement overrides

Functions

check

check(config, config_path, raw_yaml)

Check all instrument references against the loaded profile registry.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check all instrument references against the loaded profile registry."""
    # Load known instruments — gracefully degrade on failure
    try:
        from marianne.instruments.loader import load_all_profiles

        known = set(load_all_profiles().keys())
    except Exception:
        _logger.debug("V210: could not load instrument profiles, skipping check")
        return []

    if not known:
        return []

    # Score-level instrument aliases are valid references — they resolve
    # to profile names at build time via config.instruments[name].profile.
    score_instruments = set(config.instruments.keys())
    all_valid = known | score_instruments

    issues: list[ValidationIssue] = []

    # 1. Top-level instrument: field
    # Must check against all_valid (profiles + score aliases), not just
    # known (profiles only). A score can define instrument: my-alias and
    # instruments: { my-alias: { profile: claude-code } } — that's valid.
    if config.instrument and config.instrument not in all_valid:
        issues.append(self._make_issue(
            config.instrument,
            "score-level instrument",
            find_line_in_yaml(raw_yaml, "instrument:"),
            all_valid,
        ))

    # 2. Per-sheet instruments
    if config.sheet.per_sheet_instruments:
        for sheet_num, instr_name in config.sheet.per_sheet_instruments.items():
            if instr_name not in all_valid:
                issues.append(self._make_issue(
                    instr_name,
                    f"sheet {sheet_num} instrument",
                    find_line_in_yaml(raw_yaml, f"{sheet_num}:"),
                    all_valid,
                ))

    # 3. Instrument map
    if config.sheet.instrument_map:
        for instr_name in config.sheet.instrument_map:
            if instr_name not in all_valid:
                issues.append(self._make_issue(
                    instr_name,
                    "instrument_map entry",
                    find_line_in_yaml(raw_yaml, f"{instr_name}:"),
                    all_valid,
                ))

    # 4. Movement-level instruments
    if config.movements:
        for mov_num, mov_def in config.movements.items():
            if mov_def.instrument and mov_def.instrument not in all_valid:
                issues.append(self._make_issue(
                    mov_def.instrument,
                    f"movement {mov_num} instrument",
                    find_line_in_yaml(raw_yaml, f"{mov_num}:"),
                    all_valid,
                ))

    return issues

InstrumentFallbackCheck

Check that instrument_fallbacks references resolve to known profiles (V211).

Warns when a fallback instrument name doesn't match any loaded instrument profile or score-level instrument alias. Same severity as V210 — the conductor may have instruments the validator doesn't know about.

Checks: - Score-level instrument_fallbacks - movements.N.instrument_fallbacks per-movement fallbacks - sheet.per_sheet_fallbacks per-sheet fallback overrides

Functions

check

check(config, config_path, raw_yaml)

Check all instrument fallback references against the loaded profile registry.

Source code in src/marianne/validation/checks/config.py

def check(
    self,
    config: JobConfig,
    config_path: Path,
    raw_yaml: str,
) -> list[ValidationIssue]:
    """Check all instrument fallback references against the loaded profile registry."""
    try:
        from marianne.instruments.loader import load_all_profiles

        known = set(load_all_profiles().keys())
    except Exception:
        _logger.debug("V211: could not load instrument profiles, skipping check")
        return []

    if not known:
        return []

    # Score-level instrument aliases are valid fallback targets
    score_instruments = set(config.instruments.keys())
    all_valid = known | score_instruments

    issues: list[ValidationIssue] = []

    # 1. Score-level instrument_fallbacks
    for name in config.instrument_fallbacks:
        if name not in all_valid:
            issues.append(self._make_issue(
                name,
                "score-level instrument_fallbacks",
                find_line_in_yaml(raw_yaml, name),
                all_valid,
            ))

    # 2. Movement-level instrument_fallbacks
    for mov_num, mov_def in config.movements.items():
        for name in mov_def.instrument_fallbacks:
            if name not in all_valid:
                issues.append(self._make_issue(
                    name,
                    f"movement {mov_num} instrument_fallbacks",
                    find_line_in_yaml(raw_yaml, name),
                    all_valid,
                ))

    # 3. Per-sheet fallbacks
    for sheet_num, fallback_list in config.sheet.per_sheet_fallbacks.items():
        for name in fallback_list:
            if name not in all_valid:
                issues.append(self._make_issue(
                    name,
                    f"sheet {sheet_num} per_sheet_fallbacks",
                    find_line_in_yaml(raw_yaml, name),
                    all_valid,
                ))

    return issues

Functions