job

job

Job and sheet configuration models.

Defines the top-level JobConfig, SheetConfig, and PromptConfig models.

Classes

InjectionCategory

Bases: str, Enum

Category for injected content in prelude/cadenza system.

Determines WHERE in the prompt the injected content appears: - context: Background knowledge, after template body - skill: Methodology/instructions, after preamble - tool: Available actions, after preamble

InjectionItem

Bases: BaseModel

A single injection item referencing a file or directory with a category.

Used in prelude (all sheets) and cadenzas (per-sheet) to inject file content into prompts at category-appropriate locations.

Supports two mutually exclusive modes: - file: inject a single file's content - directory: inject all files in a directory (directory cadenza)

Functions

exactly_one_source

exactly_one_source()

Ensure exactly one of file or directory is specified.

Source code in src/marianne/core/config/job.py

@model_validator(mode="after")
def exactly_one_source(self) -> InjectionItem:
    """Ensure exactly one of file or directory is specified."""
    if self.file and self.directory:
        raise ValueError("Specify 'file' or 'directory', not both.")
    if not self.file and not self.directory:
        raise ValueError("One of 'file' or 'directory' is required.")
    return self

InstrumentDef

Bases: BaseModel

A named instrument definition within a score.

Allows a score to declare reusable instrument aliases that reference registered instrument profiles with optional configuration overrides. These aliases can then be referenced by name in per-sheet or per-movement instrument assignments.

Example YAML::

instruments:
  fast-writer:
    profile: gemini-cli
    config:
      model: gemini-2.5-flash
      timeout_seconds: 300
  deep-thinker:
    profile: claude-code
    config:
      timeout_seconds: 3600

MovementDef

Bases: BaseModel

Declaration of a movement within a score.

Movements are sequential execution phases. Each movement can specify a name, an instrument (overriding the score default), instrument config, and a voice count (shorthand for fan-out).

Example YAML::

movements:
  1:
    name: Planning
    instrument: claude-code
  2:
    name: Implementation
    voices: 3
    instrument: gemini-cli
  3:
    name: Review

SheetConfig

Bases: BaseModel

Configuration for sheet processing.

In Marianne's musical theme, a composition is divided into sheets, each containing a portion of the work to be performed.

Fan-out support: When fan_out is specified, stages are expanded into concrete sheets at parse time. For example, total_items=7, fan_out={2: 3} produces 9 concrete sheets (stage 2 instantiated 3 times). After expansion, total_items and dependencies reflect expanded values, and fan_out is cleared to {} to prevent re-expansion on resume.

Attributes

total_sheets property

total_sheets

Calculate total number of sheets.

total_stages property

total_stages

Return the original stage count.

After fan-out expansion, total_items reflects expanded sheet count. total_stages preserves the original logical stage count from fan_out_stage_map. When no fan-out was used, total_stages == total_sheets (identity).

Functions

strip_computed_fields classmethod

strip_computed_fields(data)

Strip computed properties that users may include in YAML.

total_sheets is computed from size/total_items, not configurable. Accept it silently for backward compatibility — rejecting it would break existing scores that include it.

Source code in src/marianne/core/config/job.py

@model_validator(mode="before")
@classmethod
def strip_computed_fields(cls, data: Any) -> Any:
    """Strip computed properties that users may include in YAML.

    total_sheets is computed from size/total_items, not configurable.
    Accept it silently for backward compatibility — rejecting it would
    break existing scores that include it.
    """
    if isinstance(data, dict) and "total_sheets" in data:
        data.pop("total_sheets")
    return data

validate_per_sheet_instruments classmethod

validate_per_sheet_instruments(v)

Validate per-sheet instrument assignments.

Source code in src/marianne/core/config/job.py

@field_validator("per_sheet_instruments")
@classmethod
def validate_per_sheet_instruments(
    cls, v: dict[int, str],
) -> dict[int, str]:
    """Validate per-sheet instrument assignments."""
    for sheet_num, instrument in v.items():
        if not isinstance(sheet_num, int) or sheet_num < 1:
            raise ValueError(
                f"Per-sheet instrument key must be a positive integer, "
                f"got {sheet_num}"
            )
        if not instrument:
            raise ValueError(
                f"Per-sheet instrument name for sheet {sheet_num} "
                f"must not be empty"
            )
    return v

validate_per_sheet_fallbacks classmethod

validate_per_sheet_fallbacks(v)

Validate per-sheet fallback chain keys are positive integers.

Source code in src/marianne/core/config/job.py

@field_validator("per_sheet_fallbacks")
@classmethod
def validate_per_sheet_fallbacks(
    cls, v: dict[int, list[str]],
) -> dict[int, list[str]]:
    """Validate per-sheet fallback chain keys are positive integers."""
    for sheet_num in v:
        if not isinstance(sheet_num, int) or sheet_num < 1:
            raise ValueError(
                f"Per-sheet fallback key must be a positive integer, "
                f"got {sheet_num}"
            )
    return v

validate_instrument_map classmethod

validate_instrument_map(v)

Validate instrument_map: no duplicate sheets, valid names.

Source code in src/marianne/core/config/job.py

@field_validator("instrument_map")
@classmethod
def validate_instrument_map(
    cls, v: dict[str, list[int]],
) -> dict[str, list[int]]:
    """Validate instrument_map: no duplicate sheets, valid names."""
    seen_sheets: dict[int, str] = {}
    for instrument, sheets in v.items():
        if not instrument:
            raise ValueError(
                "Instrument name in instrument_map must not be empty"
            )
        for sheet_num in sheets:
            if not isinstance(sheet_num, int) or sheet_num < 1:
                raise ValueError(
                    f"Sheet number in instrument_map must be a positive "
                    f"integer, got {sheet_num} for instrument '{instrument}'"
                )
            if sheet_num in seen_sheets:
                raise ValueError(
                    f"Sheet {sheet_num} assigned to multiple instruments "
                    f"in instrument_map: '{seen_sheets[sheet_num]}' and "
                    f"'{instrument}'"
                )
            seen_sheets[sheet_num] = instrument
    return v

get_fan_out_metadata

get_fan_out_metadata(sheet_num)

Get fan-out metadata for a specific sheet.

Parameters:

Name	Type	Description	Default
sheet_num	int	Concrete sheet number (1-indexed).	required

Returns:

Type	Description
FanOutMetadata	FanOutMetadata with stage, instance, and fan_count.
FanOutMetadata	When no fan-out is configured, returns identity metadata
FanOutMetadata	(stage=sheet_num, instance=1, fan_count=1).

Source code in src/marianne/core/config/job.py

def get_fan_out_metadata(self, sheet_num: int) -> FanOutMetadata:  # noqa: F821
    """Get fan-out metadata for a specific sheet.

    Args:
        sheet_num: Concrete sheet number (1-indexed).

    Returns:
        FanOutMetadata with stage, instance, and fan_count.
        When no fan-out is configured, returns identity metadata
        (stage=sheet_num, instance=1, fan_count=1).
    """
    from marianne.core.fan_out import FanOutMetadata

    if self.fan_out_stage_map and sheet_num in self.fan_out_stage_map:
        meta = self.fan_out_stage_map[sheet_num]
        return FanOutMetadata(
            stage=meta["stage"],
            instance=meta["instance"],
            fan_count=meta["fan_count"],
        )
    return FanOutMetadata(stage=sheet_num, instance=1, fan_count=1)

validate_fan_out classmethod

validate_fan_out(v)

Validate fan_out field values.

Source code in src/marianne/core/config/job.py

@field_validator("fan_out")
@classmethod
def validate_fan_out(cls, v: dict[int, int]) -> dict[int, int]:
    """Validate fan_out field values."""
    for stage, count in v.items():
        if not isinstance(stage, int) or stage < 1:
            raise ValueError(
                f"Fan-out stage must be positive integer, got {stage}"
            )
        if not isinstance(count, int) or count < 1:
            raise ValueError(
                f"Fan-out count for stage {stage} must be >= 1, got {count}"
            )
    return v

validate_dependencies classmethod

validate_dependencies(v, info)

Validate dependency declarations.

Note: Full validation (range checks, cycle detection) happens when the DependencyDAG is built at runtime, since total_sheets isn't available during field validation.

Source code in src/marianne/core/config/job.py

@field_validator("dependencies")
@classmethod
def validate_dependencies(
    cls, v: dict[int, list[int]], info: ValidationInfo
) -> dict[int, list[int]]:
    """Validate dependency declarations.

    Note: Full validation (range checks, cycle detection) happens when
    the DependencyDAG is built at runtime, since total_sheets isn't
    available during field validation.
    """
    for sheet_num, deps in v.items():
        if not isinstance(sheet_num, int) or sheet_num < 1:
            raise ValueError(f"Sheet number must be positive integer, got {sheet_num}")
        if not isinstance(deps, list):
            raise ValueError(f"Dependencies for sheet {sheet_num} must be a list")
        for dep in deps:
            if not isinstance(dep, int) or dep < 1:
                raise ValueError(
                    f"Dependency must be positive integer, got {dep} for sheet {sheet_num}"
                )
            if dep == sheet_num:
                raise ValueError(f"Sheet {sheet_num} cannot depend on itself")
    return v

expand_fan_out_config

expand_fan_out_config()

Expand fan_out declarations into concrete sheet assignments.

This runs after field validators. When fan_out is non-empty: 1. Validates constraints (size=1, start_item=1) 2. Calls expand_fan_out() to compute concrete sheet assignments 3. Overwrites total_items and dependencies with expanded values 4. Stores metadata in fan_out_stage_map for resume support 5. Clears fan_out={} to prevent re-expansion on resume

Source code in src/marianne/core/config/job.py

@model_validator(mode="after")
def expand_fan_out_config(self) -> SheetConfig:
    """Expand fan_out declarations into concrete sheet assignments.

    This runs after field validators. When fan_out is non-empty:
    1. Validates constraints (size=1, start_item=1)
    2. Calls expand_fan_out() to compute concrete sheet assignments
    3. Overwrites total_items and dependencies with expanded values
    4. Stores metadata in fan_out_stage_map for resume support
    5. Clears fan_out={} to prevent re-expansion on resume
    """
    if not self.fan_out:
        return self

    # Enforce constraints for fan-out
    if self.size != 1:
        raise ValueError(
            f"fan_out requires size=1, got size={self.size}. "
            "Each stage must map to exactly one sheet for fan-out to work."
        )
    if self.start_item != 1:
        raise ValueError(
            f"fan_out requires start_item=1, got start_item={self.start_item}. "
            "Fan-out stages are 1-indexed from the beginning."
        )

    from marianne.core.fan_out import expand_fan_out

    expansion = expand_fan_out(
        total_stages=self.total_items,
        fan_out=self.fan_out,
        stage_dependencies=self.dependencies,
    )

    # Overwrite with expanded values
    self.total_items = expansion.total_sheets
    self.dependencies = expansion.expanded_dependencies

    # Expand skip_when: stage-keyed → sheet-keyed
    if self.skip_when:
        expanded_skip_when: dict[int, str] = {}
        for stage, expr in self.skip_when.items():
            for sheet_num in expansion.stage_sheets.get(stage, [stage]):
                expanded_skip_when[sheet_num] = expr
        self.skip_when = expanded_skip_when

    # Expand skip_when_command: stage-keyed → sheet-keyed
    if self.skip_when_command:
        expanded_skip_when_command: dict[int, SkipWhenCommand] = {}
        for stage, cmd in self.skip_when_command.items():
            for sheet_num in expansion.stage_sheets.get(stage, [stage]):
                expanded_skip_when_command[sheet_num] = cmd
        self.skip_when_command = expanded_skip_when_command

    # Store serializable metadata for resume
    self.fan_out_stage_map = {
        sheet_num: {
            "stage": meta.stage,
            "instance": meta.instance,
            "fan_count": meta.fan_count,
        }
        for sheet_num, meta in expansion.sheet_metadata.items()
    }

    # Clear fan_out to prevent re-expansion on resume
    self.fan_out = {}

    return self

validate_dependency_range

validate_dependency_range()

Validate that dependency sheet numbers are within the valid range.

Runs after fan-out expansion so total_sheets reflects the final count.

Source code in src/marianne/core/config/job.py

@model_validator(mode="after")
def validate_dependency_range(self) -> SheetConfig:
    """Validate that dependency sheet numbers are within the valid range.

    Runs after fan-out expansion so total_sheets reflects the final count.
    """
    if not self.dependencies:
        return self
    max_sheet = self.total_sheets
    for sheet_num, deps in self.dependencies.items():
        if sheet_num < 1 or sheet_num > max_sheet:
            raise ValueError(
                f"Dependency key sheet {sheet_num} is out of range "
                f"(valid: 1-{max_sheet})"
            )
        for dep in deps:
            if dep < 1 or dep > max_sheet:
                raise ValueError(
                    f"Sheet {sheet_num} depends on sheet {dep}, "
                    f"which is out of range (valid: 1-{max_sheet})"
                )
    return self

PromptConfig

Bases: BaseModel

Configuration for prompt templating.

Functions

at_least_one_template

at_least_one_template()

Warn when no template source is provided (falls back to default prompt).

Source code in src/marianne/core/config/job.py

@model_validator(mode="after")
def at_least_one_template(self) -> PromptConfig:
    """Warn when no template source is provided (falls back to default prompt)."""
    if self.template is not None and self.template_file is not None:
        raise ValueError(
            "PromptConfig accepts 'template' or 'template_file', not both"
        )
    if self.template is None and self.template_file is None:
        warnings.warn(
            "PromptConfig has neither 'template' nor 'template_file'. "
            "The default preamble prompt will be used.",
            UserWarning,
            stacklevel=2,
        )
    return self

JobConfig

Bases: BaseModel

Complete configuration for an orchestration job.

Functions

to_yaml

to_yaml(*, exclude_defaults=False)

Serialize this JobConfig to valid score YAML.

The output is semantically equivalent to the original config: from_yaml_string(config.to_yaml()) produces an equivalent config (compared via model_dump()). String-level identity with the original YAML file is NOT guaranteed because workspace paths are resolved to absolute at parse time and fan-out configs are expanded.

Parameters:

Name	Type	Description	Default
exclude_defaults	bool	If True, omit fields that match their default values for cleaner output. Defaults to False (lossless).	False

Returns:

Type	Description
str	A valid YAML string that from_yaml_string() can parse.

Source code in src/marianne/core/config/job.py

def to_yaml(self, *, exclude_defaults: bool = False) -> str:
    """Serialize this JobConfig to valid score YAML.

    The output is semantically equivalent to the original config:
    ``from_yaml_string(config.to_yaml())`` produces an equivalent config
    (compared via ``model_dump()``). String-level identity with the
    original YAML file is NOT guaranteed because workspace paths are
    resolved to absolute at parse time and fan-out configs are expanded.

    Args:
        exclude_defaults: If True, omit fields that match their default
            values for cleaner output. Defaults to False (lossless).

    Returns:
        A valid YAML string that ``from_yaml_string()`` can parse.
    """
    data = self.model_dump(
        mode="python",
        by_alias=True,
        exclude_defaults=exclude_defaults,
    )
    data = _prepare_for_yaml(data)
    return yaml.dump(
        data,
        default_flow_style=False,
        sort_keys=False,
        allow_unicode=True,
    )

from_yaml classmethod

from_yaml(path)

Load job configuration from a YAML file.

Source code in src/marianne/core/config/job.py

@classmethod
def from_yaml(cls, path: Path) -> JobConfig:
    """Load job configuration from a YAML file."""
    with open(path) as f:
        data = yaml.safe_load(f)
    if not isinstance(data, dict):
        raise ValueError(
            "The score file is empty or invalid. "
            "A Marianne score requires at minimum: name, sheet, and prompt sections. "
            "See 'mzt validate --help' or the score writing guide for examples."
        )
    # Pre-resolve relative workspace relative to the score file's parent
    # directory, not the current process CWD (#109).  This is critical when
    # the daemon loads a score whose path differs from the daemon's CWD.
    if "workspace" in data:
        ws = Path(str(data["workspace"]))
        if not ws.is_absolute():
            data["workspace"] = str((path.resolve().parent / ws).resolve())
    return cls.model_validate(data)

from_yaml_string classmethod

from_yaml_string(yaml_str)

Load job configuration from a YAML string.

Source code in src/marianne/core/config/job.py

@classmethod
def from_yaml_string(cls, yaml_str: str) -> JobConfig:
    """Load job configuration from a YAML string."""
    data = yaml.safe_load(yaml_str)
    if not isinstance(data, dict):
        raise ValueError(
            "The score content is empty or invalid. "
            "A Marianne score requires at minimum: name, sheet, and prompt sections."
        )
    return cls.model_validate(data)

get_state_path

get_state_path()

Get the resolved state path.

Source code in src/marianne/core/config/job.py

def get_state_path(self) -> Path:
    """Get the resolved state path."""
    if self.state_path:
        return self.state_path
    if self.state_backend == "json":
        return self.workspace / ".marianne-state.json"
    return self.workspace / ".marianne-state.db"

get_outcome_store_path

get_outcome_store_path()

Get the resolved outcome store path for learning.

Source code in src/marianne/core/config/job.py

def get_outcome_store_path(self) -> Path:
    """Get the resolved outcome store path for learning."""
    if self.learning.outcome_store_path:
        return self.learning.outcome_store_path
    if self.learning.outcome_store_type == "json":
        return self.workspace / ".marianne-outcomes.json"
    return self.workspace / ".marianne-outcomes.db"