Index

compose

Composition compiler — turns semantic agent definitions into Mozart scores.

The compiler takes high-level descriptions (agent identities, patterns, techniques, instrument assignments) and produces complete Mozart score YAML.

Classes

FleetGenerator

Generates fleet configuration from a compiler config.

Takes the full compiler config with agent roster and produces a fleet YAML that references individual agent scores with group dependencies for startup ordering.

Functions

generate

generate(config, scores_dir)

Generate a fleet config from the compiler config.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Full compiler config with project, agents, groups.	required
scores_dir	Path	Directory where individual agent scores will be written.	required

Returns:

Type	Description
dict[str, Any]	Fleet config dict suitable for YAML serialization.

Source code in src/marianne/compose/fleet.py

def generate(
    self,
    config: dict[str, Any],
    scores_dir: Path,
) -> dict[str, Any]:
    """Generate a fleet config from the compiler config.

    Args:
        config: Full compiler config with project, agents, groups.
        scores_dir: Directory where individual agent scores will be written.

    Returns:
        Fleet config dict suitable for YAML serialization.
    """
    project_name = config.get("project", {}).get("name", "unnamed")
    agents = config.get("agents", [])

    if not agents:
        raise ValueError("Fleet config requires at least one agent")

    # Build score entries
    scores: list[dict[str, str | None]] = []
    for agent in agents:
        name = agent.get("name", "")
        if not name:
            continue
        group = agent.get("group")
        entry: dict[str, str | None] = {
            "path": str(scores_dir / f"{name}.yaml"),
        }
        if group:
            entry["group"] = group
        scores.append(entry)

    # Build group definitions from config
    groups = self._build_groups(config)

    fleet: dict[str, Any] = {
        "name": f"{project_name}-fleet",
        "type": "fleet",
        "scores": scores,
    }

    if groups:
        fleet["groups"] = groups

    return fleet

write

write(config, scores_dir, output_path)

Generate and write a fleet config to disk.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Full compiler config.	required
scores_dir	Path	Directory where agent scores live.	required
output_path	Path	Path to write the fleet YAML.	required

Returns:

Type	Description
Path	Path to the written fleet config.

Source code in src/marianne/compose/fleet.py

def write(
    self,
    config: dict[str, Any],
    scores_dir: Path,
    output_path: Path,
) -> Path:
    """Generate and write a fleet config to disk.

    Args:
        config: Full compiler config.
        scores_dir: Directory where agent scores live.
        output_path: Path to write the fleet YAML.

    Returns:
        Path to the written fleet config.
    """
    fleet = self.generate(config, scores_dir)

    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        yaml.dump(fleet, f, default_flow_style=False, sort_keys=False)

    _logger.info("Fleet config written to %s", output_path)
    return output_path

IdentitySeeder

IdentitySeeder(agents_dir=None)

Creates the L1-L4 identity stack for agents.

The seeder is idempotent: running it on an existing agent directory updates files without corrupting existing identity data. Existing content in L3 (recent) and L4 (growth) is preserved if present.

Source code in src/marianne/compose/identity.py

def __init__(self, agents_dir: Path | None = None) -> None:
    self.agents_dir = agents_dir or DEFAULT_AGENTS_DIR

Functions

seed

seed(agent_def, *, existing_memory_path=None, existing_meditation_path=None)

Create the full identity store for an agent.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition dict with keys: name, voice, focus, and optionally: role, meditation, a2a_skills, techniques.	required
existing_memory_path	Path | None	Path to existing memory file for migration (distilled into L3 recent.md).	None
existing_meditation_path	Path | None	Path to existing meditation file for migration (distilled into L1 stakes/identity).	None

Returns:

Type	Description
Path	Path to the agent's identity directory.

Raises:

Type	Description
ValueError	If agent_def is missing required fields.

Source code in src/marianne/compose/identity.py

def seed(
    self,
    agent_def: dict[str, Any],
    *,
    existing_memory_path: Path | None = None,
    existing_meditation_path: Path | None = None,
) -> Path:
    """Create the full identity store for an agent.

    Args:
        agent_def: Agent definition dict with keys: name, voice, focus,
            and optionally: role, meditation, a2a_skills, techniques.
        existing_memory_path: Path to existing memory file for migration
            (distilled into L3 recent.md).
        existing_meditation_path: Path to existing meditation file for
            migration (distilled into L1 stakes/identity).

    Returns:
        Path to the agent's identity directory.

    Raises:
        ValueError: If agent_def is missing required fields.
    """
    name = agent_def.get("name")
    if not name:
        raise ValueError("Agent definition must include 'name'")
    voice = agent_def.get("voice", "")
    focus = agent_def.get("focus", "")

    if not voice:
        raise ValueError(f"Agent '{name}' must have a 'voice'")
    if not focus:
        raise ValueError(f"Agent '{name}' must have a 'focus'")

    agent_dir: Path = self.agents_dir / str(name)
    agent_dir.mkdir(parents=True, exist_ok=True)
    archive_dir = agent_dir / "archive"
    archive_dir.mkdir(exist_ok=True)

    self._create_identity_md(agent_dir, agent_def, existing_meditation_path)
    self._create_profile_yaml(agent_dir, agent_def)
    self._create_recent_md(agent_dir, agent_def, existing_memory_path)
    self._create_growth_md(agent_dir, agent_def)

    _logger.info("Agent '%s' identity seeded at %s", name, agent_dir)
    return agent_dir

seed_all

seed_all(agents, *, migration_memory_dir=None, migration_meditation_dir=None)

Seed identity for all agents in a roster.

Parameters:

Name	Type	Description	Default
agents	list[dict[str, Any]]	List of agent definition dicts.	required
migration_memory_dir	Path | None	Directory containing existing memory files named {agent_name}.md for migration.	None
migration_meditation_dir	Path | None	Directory containing existing meditation files named {agent_name}.md for migration.	None

Returns:

Type	Description
list[Path]	List of paths to agent identity directories.

Source code in src/marianne/compose/identity.py

def seed_all(
    self,
    agents: list[dict[str, Any]],
    *,
    migration_memory_dir: Path | None = None,
    migration_meditation_dir: Path | None = None,
) -> list[Path]:
    """Seed identity for all agents in a roster.

    Args:
        agents: List of agent definition dicts.
        migration_memory_dir: Directory containing existing memory files
            named ``{agent_name}.md`` for migration.
        migration_meditation_dir: Directory containing existing meditation
            files named ``{agent_name}.md`` for migration.

    Returns:
        List of paths to agent identity directories.
    """
    results: list[Path] = []
    for agent_def in agents:
        name = agent_def.get("name", "")
        memory_path = None
        meditation_path = None

        if migration_memory_dir and name:
            candidate = migration_memory_dir / f"{name}.md"
            if candidate.exists():
                memory_path = candidate

        if migration_meditation_dir and name:
            candidate = migration_meditation_dir / f"{name}.md"
            if candidate.exists():
                meditation_path = candidate

        path = self.seed(
            agent_def,
            existing_memory_path=memory_path,
            existing_meditation_path=meditation_path,
        )
        results.append(path)
    return results

InstrumentResolver

Resolves per-sheet instrument assignments with deep fallback chains.

Produces a matrix of primary instruments and fallback chains for every sheet in the cycle. Free-tier models are the defaults; paid models are power-ups available in the fallback chain.

Functions

resolve

resolve(agent_def, defaults)

Resolve instrument assignments for an agent.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition with optional instrument overrides.	required
defaults	dict[str, Any]	Global defaults with instrument definitions per tier.	required

Returns:

Type	Description
dict[str, Any]	Dict with keys: backend: dict — primary backend config instrument_fallbacks: list[str] — score-level fallbacks per_sheet_instruments: dict[int, str] — per-sheet primary per_sheet_instrument_config: dict[int, dict] — per-sheet config per_sheet_fallbacks: dict[int, list[str]] — per-sheet fallback chains

Source code in src/marianne/compose/instruments.py

def resolve(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
) -> dict[str, Any]:
    """Resolve instrument assignments for an agent.

    Args:
        agent_def: Agent definition with optional instrument overrides.
        defaults: Global defaults with instrument definitions per tier.

    Returns:
        Dict with keys:
            ``backend``: dict — primary backend config
            ``instrument_fallbacks``: list[str] — score-level fallbacks
            ``per_sheet_instruments``: dict[int, str] — per-sheet primary
            ``per_sheet_instrument_config``: dict[int, dict] — per-sheet config
            ``per_sheet_fallbacks``: dict[int, list[str]] — per-sheet fallback chains
    """
    default_instruments = defaults.get("instruments", {})
    agent_instruments = agent_def.get("instruments", {})

    per_sheet_instruments: dict[int, str] = {}
    per_sheet_config: dict[int, dict[str, Any]] = {}
    per_sheet_fallbacks: dict[int, list[str]] = {}

    # Build the full instrument catalog for tail fallbacks
    all_instruments = self._collect_all_instruments(default_instruments)

    for sheet_num in range(1, SHEETS_PER_CYCLE + 1):
        phase = SHEET_PHASE.get(sheet_num, "work")
        tier = PHASE_TIER_MAP.get(phase, "work")

        # Resolve: agent override > default for this tier
        resolved = self._resolve_for_tier(
            tier, agent_instruments, default_instruments
        )

        if resolved:
            primary = resolved.get("primary", {})
            instrument_name = primary.get("instrument", "")
            if instrument_name:
                per_sheet_instruments[sheet_num] = instrument_name

            model = primary.get("model", "")
            provider = primary.get("provider", "")
            timeout = primary.get("timeout_seconds", 0)
            config: dict[str, Any] = {}
            if model:
                config["model"] = model
            if provider:
                config["provider"] = provider
            if timeout:
                config["timeout_seconds"] = timeout
            if config:
                per_sheet_config[sheet_num] = config

            # Build fallback chain: explicit fallbacks + full catalog tail
            fallbacks = self._build_fallback_chain(
                resolved.get("fallbacks", []),
                all_instruments,
                exclude=instrument_name,
            )
            if fallbacks:
                per_sheet_fallbacks[sheet_num] = fallbacks

    # Determine score-level primary backend
    work_tier = self._resolve_for_tier(
        "work", agent_instruments, default_instruments
    )
    primary_backend = self._to_backend_config(work_tier)

    # Score-level fallbacks from the full catalog
    score_fallbacks = list(all_instruments)

    return {
        "backend": primary_backend,
        "instrument_fallbacks": score_fallbacks,
        "per_sheet_instruments": per_sheet_instruments,
        "per_sheet_instrument_config": per_sheet_config,
        "per_sheet_fallbacks": per_sheet_fallbacks,
    }

PatternExpander

PatternExpander(custom_patterns=None)

Expands named patterns into sheet sequence modifiers.

Patterns inject additional prompt context and structural guidance into specific phases of the agent cycle. Two expansion modes:

• expand() — prompt extensions (per-phase text additions).
• expand_stages() — concrete stage sequence with purposes, instrument guidance, and validation shapes.

Initialize with optional custom patterns.

Parameters:

Name	Type	Description	Default
custom_patterns	dict[str, dict[str, Any]] | None	Additional patterns to register beyond builtins.	None

Source code in src/marianne/compose/patterns.py

def __init__(self, custom_patterns: dict[str, dict[str, Any]] | None = None) -> None:
    """Initialize with optional custom patterns.

    Args:
        custom_patterns: Additional patterns to register beyond builtins.
    """
    self.patterns: dict[str, dict[str, Any]] = dict(BUILTIN_PATTERNS)
    self._stage_defs: dict[str, dict[str, Any]] = dict(_BUILTIN_STAGE_DEFS)
    if custom_patterns:
        self.patterns.update(custom_patterns)

Functions

expand

expand(pattern_names, agent_def)

Expand named patterns into per-phase prompt extensions.

Parameters:

Name	Type	Description	Default
pattern_names	list[str]	List of pattern names to apply.	required
agent_def	dict[str, Any]	Agent definition for context.	required

Returns:

Type	Description
dict[str, Any]	Dict with keys: prompt_extensions: dict[str, str] — per-phase prompt additions applied_patterns: list[str] — names of patterns actually applied

Raises:

Type	Description
ValueError	If a pattern name is not recognised.

Source code in src/marianne/compose/patterns.py

def expand(
    self,
    pattern_names: list[str],
    agent_def: dict[str, Any],
) -> dict[str, Any]:
    """Expand named patterns into per-phase prompt extensions.

    Args:
        pattern_names: List of pattern names to apply.
        agent_def: Agent definition for context.

    Returns:
        Dict with keys:
            ``prompt_extensions``: dict[str, str] — per-phase prompt additions
            ``applied_patterns``: list[str] — names of patterns actually applied

    Raises:
        ValueError: If a pattern name is not recognised.
    """
    prompt_extensions: dict[str, str] = {}
    applied: list[str] = []

    for name in pattern_names:
        pattern = self.patterns.get(name)
        if pattern is None:
            raise ValueError(
                f"Unknown pattern '{name}'. "
                f"Available: {sorted(self.patterns.keys())}"
            )

        modifiers = pattern.get("sheet_modifiers", {})
        for phase, mods in modifiers.items():
            if isinstance(mods, dict):
                extension = mods.get("prompt_extension", "")
                if extension:
                    if phase in prompt_extensions:
                        prompt_extensions[phase] += f"\n{extension}"
                    else:
                        prompt_extensions[phase] = extension

        applied.append(name)
        _logger.debug(
            "Applied pattern '%s' to agent '%s'",
            name, agent_def.get("name"),
        )

    return {
        "prompt_extensions": prompt_extensions,
        "applied_patterns": applied,
    }

expand_stages

expand_stages(pattern_name, params=None)

Expand a pattern into a concrete stage sequence.

Parameters:

Name	Type	Description	Default
pattern_name	str	Name of the pattern to expand.	required
params	dict[str, Any] | None	Optional parameters to customise expansion. Supported keys: fan_out — dict mapping stage name to fan-out width (overrides the pattern default).	None

Returns:

Type	Description
list[PatternStage]	Ordered list of :class:PatternStage instances.

Raises:

Type	Description
ValueError	If the pattern is unknown or has no stage definition.

Source code in src/marianne/compose/patterns.py

def expand_stages(
    self,
    pattern_name: str,
    params: dict[str, Any] | None = None,
) -> list[PatternStage]:
    """Expand a pattern into a concrete stage sequence.

    Args:
        pattern_name: Name of the pattern to expand.
        params: Optional parameters to customise expansion.
            Supported keys:
                ``fan_out`` — dict mapping stage name to fan-out width
                    (overrides the pattern default).

    Returns:
        Ordered list of :class:`PatternStage` instances.

    Raises:
        ValueError: If the pattern is unknown or has no stage definition.
    """
    stage_def = self._stage_defs.get(pattern_name)
    if stage_def is None:
        available = sorted(self._stage_defs.keys())
        raise ValueError(
            f"Unknown pattern '{pattern_name}' (or pattern has no stage "
            f"definition). Available: {available}"
        )

    # Merge fan_out overrides
    default_fan_out: dict[str, int] = dict(stage_def.get("fan_out", {}))
    if params and "fan_out" in params:
        default_fan_out.update(params["fan_out"])

    stages: list[PatternStage] = []
    for raw in stage_def["stages"]:
        sheets_val = raw["sheets"]

        # Apply fan_out parameter override if applicable
        if isinstance(sheets_val, str) and sheets_val.startswith("fan_out("):
            stage_name = raw["name"]
            if stage_name in default_fan_out:
                sheets_val = f"fan_out({default_fan_out[stage_name]})"
        elif isinstance(sheets_val, int) and raw["name"] in default_fan_out:
            sheets_val = f"fan_out({default_fan_out[raw['name']]})"

        artifacts = raw.get("artifacts", [])
        stages.append(
            PatternStage(
                name=raw["name"],
                sheets=sheets_val,
                purpose=raw.get("purpose", ""),
                instrument_guidance=raw.get("instrument_guidance", ""),
                fallback_friendly=raw.get("fallback_friendly", True),
                artifacts=tuple(artifacts),
            )
        )

    _logger.debug(
        "Expanded pattern '%s' into %d stages", pattern_name, len(stages),
    )
    return stages

expand_stages_with_validations

expand_stages_with_validations(pattern_name, params=None)

Expand a pattern into stages and generate validation shapes.

Validation shapes are derived from each stage's declared artifacts. Each artifact produces a file_exists validation template.

Parameters:

Name	Type	Description	Default
pattern_name	str	Name of the pattern to expand.	required
params	dict[str, Any] | None	Optional expansion parameters (see :meth:expand_stages).	None

Returns:

Type	Description
dict[str, Any]	Dict with keys: stages: list[:class:PatternStage] validations: list[dict] — validation rule shapes

Source code in src/marianne/compose/patterns.py

def expand_stages_with_validations(
    self,
    pattern_name: str,
    params: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Expand a pattern into stages and generate validation shapes.

    Validation shapes are derived from each stage's declared artifacts.
    Each artifact produces a ``file_exists`` validation template.

    Args:
        pattern_name: Name of the pattern to expand.
        params: Optional expansion parameters (see :meth:`expand_stages`).

    Returns:
        Dict with keys:
            ``stages``: list[:class:`PatternStage`]
            ``validations``: list[dict] — validation rule shapes
    """
    stages = self.expand_stages(pattern_name, params=params)
    validations: list[dict[str, str]] = []

    for stage in stages:
        for artifact in stage.artifacts:
            validations.append({
                "type": "file_exists",
                "path": f"{{{{workspace}}}}/{artifact}",
                "stage": stage.name,
            })

    return {"stages": stages, "validations": validations}

list_patterns

list_patterns()

List all available patterns with descriptions.

Returns:

Type	Description
list[dict[str, str]]	List of dicts with name and description keys.

Source code in src/marianne/compose/patterns.py

def list_patterns(self) -> list[dict[str, str]]:
    """List all available patterns with descriptions.

    Returns:
        List of dicts with ``name`` and ``description`` keys.
    """
    return [
        {"name": name, "description": pattern.get("description", "")}
        for name, pattern in sorted(self.patterns.items())
    ]

get_pattern

get_pattern(name)

Get a pattern definition by name.

Returns:

Type	Description
dict[str, Any] | None	Pattern dict or None if not found.

Source code in src/marianne/compose/patterns.py

def get_pattern(self, name: str) -> dict[str, Any] | None:
    """Get a pattern definition by name.

    Returns:
        Pattern dict or ``None`` if not found.
    """
    return self.patterns.get(name)

load_rosetta_corpus

load_rosetta_corpus(corpus_dir)

Load patterns from a Rosetta corpus directory.

Reads *.md files in corpus_dir, parses YAML frontmatter, and registers patterns that contain a stages field.

Parameters:

Name	Type	Description	Default
corpus_dir	str | Path	Directory containing pattern markdown files.	required

Returns:

Type	Description
dict[str, dict[str, Any]]	Dict of loaded pattern slug → pattern data. Only patterns
dict[str, dict[str, Any]]	with parseable frontmatter and a name field are returned.

Source code in src/marianne/compose/patterns.py

def load_rosetta_corpus(
    self,
    corpus_dir: str | Path,
) -> dict[str, dict[str, Any]]:
    """Load patterns from a Rosetta corpus directory.

    Reads ``*.md`` files in *corpus_dir*, parses YAML frontmatter,
    and registers patterns that contain a ``stages`` field.

    Args:
        corpus_dir: Directory containing pattern markdown files.

    Returns:
        Dict of loaded pattern slug → pattern data.  Only patterns
        with parseable frontmatter and a ``name`` field are returned.
    """
    corpus_dir = Path(corpus_dir)
    if not corpus_dir.is_dir():
        _logger.warning("Rosetta corpus directory not found: %s", corpus_dir)
        return {}

    loaded: dict[str, dict[str, Any]] = {}

    for md_file in sorted(corpus_dir.glob("*.md")):
        try:
            text = md_file.read_text(encoding="utf-8")
        except OSError:
            _logger.warning("Could not read corpus file: %s", md_file)
            continue

        fm = _parse_frontmatter(text)
        if not fm or "name" not in fm:
            continue

        slug = _slugify(fm["name"])

        # Build a pattern entry
        pattern_entry: dict[str, Any] = {
            "description": fm.get("problem", fm.get("name", "")),
            "scale": fm.get("scale", ""),
            "status": fm.get("status", ""),
            "signals": fm.get("signals", []),
            "composes_with": fm.get("composes_with", []),
        }

        # If the pattern has stages, register them
        raw_stages = fm.get("stages", [])
        if raw_stages:
            pattern_entry["stages"] = raw_stages
            # Also register as a stage definition
            self._stage_defs.setdefault(slug, {
                "fan_out": fm.get("fan_out", {}),
                "stages": raw_stages,
            })

        # Generate sheet_modifiers from stages for prompt-extension compat
        if raw_stages and "sheet_modifiers" not in pattern_entry:
            modifiers: dict[str, dict[str, str]] = {}
            for stage in raw_stages:
                stage_name = stage.get("name", "")
                purpose = stage.get("purpose", "")
                if stage_name and purpose:
                    modifiers[stage_name] = {"prompt_extension": purpose}
            if modifiers:
                pattern_entry["sheet_modifiers"] = modifiers

        self.patterns.setdefault(slug, pattern_entry)
        loaded[slug] = pattern_entry

    _logger.info(
        "Loaded %d patterns from Rosetta corpus: %s",
        len(loaded), corpus_dir,
    )
    return loaded

PatternStage dataclass

PatternStage(name, sheets, purpose, instrument_guidance, fallback_friendly, artifacts)

A single stage within an expanded pattern.

Attributes:

Name	Type	Description
name	str	Stage identifier (e.g. recon, synthesize).
sheets	int | str	Number of sheets — an int for fixed stages, or a string like "fan_out(6)" for parameterised fan-outs.
purpose	str	What this stage accomplishes.
instrument_guidance	str	Advice on which instrument fits the stage.
fallback_friendly	bool	Whether cheaper fallback instruments are viable.
artifacts	tuple[str, ...]	Tuple of expected output file paths/globs.

CompilationPipeline

CompilationPipeline(*, agents_dir=None, techniques_dir=None, templates_dir=None)

Top-level compilation pipeline.

Coordinates all compiler modules to produce complete Mozart scores from a semantic agent configuration.

Usage::

pipeline = CompilationPipeline()
scores = pipeline.compile("config.yaml")
# Returns: list of score file paths + identity directories created

# Or programmatically:
pipeline.compile_agent(agent_def, defaults, output_dir)
pipeline.seed_identity(agent_def, agents_dir)

Source code in src/marianne/compose/pipeline.py

def __init__(
    self,
    *,
    agents_dir: Path | None = None,
    techniques_dir: Path | None = None,
    templates_dir: Path | None = None,
) -> None:
    self.identity_seeder = IdentitySeeder(agents_dir)
    self.sheet_composer = SheetComposer(templates_dir)
    self.technique_wirer = TechniqueWirer(techniques_dir)
    self.instrument_resolver = InstrumentResolver()
    self.validation_generator = ValidationGenerator()
    self.pattern_expander = PatternExpander()
    self.fleet_generator = FleetGenerator()

Functions

compile

compile(config_path, output_dir=None)

Compile a config file into Mozart scores.

Parameters:

Name	Type	Description	Default
config_path	str | Path	Path to the semantic agent config YAML.	required
output_dir	str | Path | None	Output directory for generated scores. Defaults to scores/ next to the config file.	None

Returns:

Type	Description
list[Path]	List of generated score file paths.

Source code in src/marianne/compose/pipeline.py

def compile(
    self,
    config_path: str | Path,
    output_dir: str | Path | None = None,
) -> list[Path]:
    """Compile a config file into Mozart scores.

    Args:
        config_path: Path to the semantic agent config YAML.
        output_dir: Output directory for generated scores. Defaults
            to ``scores/`` next to the config file.

    Returns:
        List of generated score file paths.
    """
    config_path = Path(config_path)
    with open(config_path) as f:
        config = yaml.safe_load(f) or {}

    if output_dir is None:
        output_dir = config_path.parent / "scores"
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    return self.compile_config(config, output_dir)

compile_config

compile_config(config, output_dir)

Compile a config dict into Mozart scores.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Parsed compiler config dict.	required
output_dir	Path	Directory to write generated scores.	required

Returns:

Type	Description
list[Path]	List of generated score file paths.

Source code in src/marianne/compose/pipeline.py

def compile_config(
    self,
    config: dict[str, Any],
    output_dir: Path,
) -> list[Path]:
    """Compile a config dict into Mozart scores.

    Args:
        config: Parsed compiler config dict.
        output_dir: Directory to write generated scores.

    Returns:
        List of generated score file paths.
    """
    agents = config.get("agents", [])
    if not agents:
        raise ValueError("Config must contain at least one agent")

    defaults = config.get("defaults", {})
    project = config.get("project", {})
    workspace = project.get("workspace", str(output_dir / "workspace"))

    score_paths: list[Path] = []

    for agent_def in agents:
        path = self.compile_agent(
            agent_def, defaults, output_dir, workspace=workspace
        )
        score_paths.append(path)

    # Generate fleet config if multiple agents
    if len(agents) > 1:
        fleet_path = output_dir / "fleet.yaml"
        self.fleet_generator.write(config, output_dir, fleet_path)
        score_paths.append(fleet_path)
        _logger.info("Fleet config written: %s", fleet_path)

    _logger.info(
        "Compiled %d agent scores to %s", len(agents), output_dir
    )
    return score_paths

compile_agent

compile_agent(agent_def, defaults, output_dir, *, workspace='')

Compile a single agent definition into a Mozart score.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition dict.	required
defaults	dict[str, Any]	Global defaults from compiler config.	required
output_dir	Path	Directory to write the score.	required
workspace	str	Workspace path for the agent.	''

Returns:

Type	Description
Path	Path to the generated score file.

Source code in src/marianne/compose/pipeline.py

def compile_agent(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
    output_dir: Path,
    *,
    workspace: str = "",
) -> Path:
    """Compile a single agent definition into a Mozart score.

    Args:
        agent_def: Agent definition dict.
        defaults: Global defaults from compiler config.
        output_dir: Directory to write the score.
        workspace: Workspace path for the agent.

    Returns:
        Path to the generated score file.
    """
    name = agent_def["name"]
    agents_dir = self.identity_seeder.agents_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    # 1. Seed identity
    self.identity_seeder.seed(agent_def)

    # 2. Compose sheet structure
    sheet_config = self.sheet_composer.compose(
        agent_def, defaults, agents_dir=agents_dir
    )

    # 3. Wire techniques
    technique_result = self.technique_wirer.wire(
        agent_def, defaults, workspace=workspace
    )

    # Merge technique cadenzas into sheet cadenzas
    for sheet_num, tech_cadenzas in technique_result["cadenzas"].items():
        if sheet_num not in sheet_config.get("cadenzas", {}):
            sheet_config.setdefault("cadenzas", {})[sheet_num] = []
        sheet_config["cadenzas"][sheet_num].extend(tech_cadenzas)

    # 4. Resolve instruments
    instrument_result = self.instrument_resolver.resolve(agent_def, defaults)

    # Merge instrument assignments into sheet config
    if instrument_result.get("per_sheet_instruments"):
        sheet_config["per_sheet_instruments"] = instrument_result["per_sheet_instruments"]
    if instrument_result.get("per_sheet_instrument_config"):
        sheet_config["per_sheet_instrument_config"] = instrument_result[
            "per_sheet_instrument_config"
        ]
    if instrument_result.get("per_sheet_fallbacks"):
        sheet_config["per_sheet_fallbacks"] = instrument_result["per_sheet_fallbacks"]

    # 5. Generate validations
    validations = self.validation_generator.generate(
        agent_def, defaults, agents_dir=str(agents_dir)
    )

    # 6. Expand patterns
    pattern_names = agent_def.get("patterns", [])
    if pattern_names:
        self.pattern_expander.expand(pattern_names, agent_def)

    # 7. Build prompt config
    prompt_config = self._build_prompt(agent_def, defaults)

    # 8. Assemble score
    score = self._assemble_score(
        name=name,
        workspace=workspace or str(output_dir / "workspace"),
        sheet_config=sheet_config,
        prompt_config=prompt_config,
        instrument_result=instrument_result,
        validations=validations,
        defaults=defaults,
    )

    # Write score
    score_path = output_dir / f"{name}.yaml"
    with open(score_path, "w") as f:
        yaml.dump(score, f, default_flow_style=False, sort_keys=False, width=120)

    # Write agent card sidecar if A2A skills declared
    agent_card = technique_result.get("agent_card")
    if agent_card:
        card_path = output_dir / f"{name}.agent-card.yaml"
        with open(card_path, "w") as f:
            yaml.dump(agent_card, f, default_flow_style=False, sort_keys=False)

    _logger.info("Score written: %s", score_path)
    return score_path

seed_identity

seed_identity(agent_def, agents_dir=None)

Seed identity for a single agent.

Convenience method that delegates to IdentitySeeder.

Source code in src/marianne/compose/pipeline.py

def seed_identity(
    self,
    agent_def: dict[str, Any],
    agents_dir: Path | None = None,
) -> Path:
    """Seed identity for a single agent.

    Convenience method that delegates to IdentitySeeder.
    """
    if agents_dir:
        seeder = IdentitySeeder(agents_dir)
    else:
        seeder = self.identity_seeder
    return seeder.seed(agent_def)

resolve_instruments

resolve_instruments(agent_def, defaults)

Resolve instruments for a single agent.

Convenience method that delegates to InstrumentResolver.

Source code in src/marianne/compose/pipeline.py

def resolve_instruments(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
) -> dict[str, Any]:
    """Resolve instruments for a single agent.

    Convenience method that delegates to InstrumentResolver.
    """
    return self.instrument_resolver.resolve(agent_def, defaults)

SheetComposer

SheetComposer(templates_dir=None)

Composes the sheet structure for an agent score.

Takes agent definitions and default config, produces the sheet section of a Mozart score YAML including fan-out, dependencies, cadenzas, and instrument assignments.

Source code in src/marianne/compose/sheets.py

def __init__(self, templates_dir: Path | None = None) -> None:
    self.templates_dir = templates_dir

Functions

compose

compose(agent_def, defaults, *, agents_dir=None)

Compose the sheet configuration for an agent score.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition dict.	required
defaults	dict[str, Any]	Global defaults from the compiler config.	required
agents_dir	Path | None	Path to agents identity directory.	None

Returns:

Type	Description
dict[str, Any]	Dict representing the sheet: section of a Mozart score.

Source code in src/marianne/compose/sheets.py

def compose(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
    *,
    agents_dir: Path | None = None,
) -> dict[str, Any]:
    """Compose the sheet configuration for an agent score.

    Args:
        agent_def: Agent definition dict.
        defaults: Global defaults from the compiler config.
        agents_dir: Path to agents identity directory.

    Returns:
        Dict representing the ``sheet:`` section of a Mozart score.
    """
    name = agent_def["name"]
    agents_dir_path = agents_dir or Path.home() / ".mzt" / "agents"
    identity_dir = str(agents_dir_path / name)

    sheet_config: dict[str, Any] = {
        "size": 1,
        "total_items": SHEETS_PER_CYCLE,
        "descriptions": dict(SHEET_DESCRIPTIONS),
        "prelude": self._build_prelude(identity_dir, defaults),
        "cadenzas": self._build_cadenzas(identity_dir, defaults),
        "fan_out": {
            # Phase 2: 3 parallel instances (integration, play, inspect)
            5: 3,
            # Phase 3: 3 parallel instances (aar, consolidate, reflect)
            8: 3,
        },
        "dependencies": self._build_dependencies(),
    }

    # Add skip_when_command for play gating
    skip_when = self._build_skip_when(agent_def, defaults, identity_dir)
    if skip_when:
        sheet_config["skip_when_command"] = skip_when

    return sheet_config

get_phase_for_sheet

get_phase_for_sheet(sheet_num)

Return the phase name for a given sheet number.

Source code in src/marianne/compose/sheets.py

def get_phase_for_sheet(self, sheet_num: int) -> str:
    """Return the phase name for a given sheet number."""
    return SHEET_PHASE.get(sheet_num, "unknown")

get_sheets_for_phase

get_sheets_for_phase(phase)

Return sheet numbers for a given phase name.

Source code in src/marianne/compose/sheets.py

def get_sheets_for_phase(self, phase: str) -> list[int]:
    """Return sheet numbers for a given phase name."""
    return PHASE_MAP.get(phase, [])

is_cli_sheet

is_cli_sheet(sheet_num)

Return True if the sheet is a CLI instrument (not an LLM call).

Source code in src/marianne/compose/sheets.py

def is_cli_sheet(self, sheet_num: int) -> bool:
    """Return True if the sheet is a CLI instrument (not an LLM call)."""
    return sheet_num in CLI_SHEETS

TechniqueWirer

TechniqueWirer(techniques_dir=None)

Wires technique declarations into per-sheet cadenza context.

Reads agent technique configs and produces: 1. Technique manifests (markdown) per phase 2. Per-sheet cadenza injections referencing technique docs 3. A2A agent card config for registration

Initialize the technique wirer.

Parameters:

Name	Type	Description	Default
techniques_dir	Path | None	Directory containing technique module documents. Falls back to searching common locations if not specified.	None

Source code in src/marianne/compose/techniques.py

def __init__(self, techniques_dir: Path | None = None) -> None:
    """Initialize the technique wirer.

    Args:
        techniques_dir: Directory containing technique module documents.
            Falls back to searching common locations if not specified.
    """
    self.techniques_dir = techniques_dir

Functions

wire

wire(agent_def, defaults, *, workspace='')

Wire techniques for an agent, returning cadenza additions and A2A config.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition with optional techniques and a2a_skills fields.	required
defaults	dict[str, Any]	Global defaults with technique declarations.	required
workspace	str	Workspace path for manifest output.	''

Returns:

Type	Description
dict[str, Any]	Dict with keys: cadenzas: dict[int, list[dict]] — per-sheet cadenza additions agent_card: dict | None — A2A agent card if a2a_skills defined technique_manifests: dict[int, str] — per-sheet manifest text

Source code in src/marianne/compose/techniques.py

def wire(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
    *,
    workspace: str = "",
) -> dict[str, Any]:
    """Wire techniques for an agent, returning cadenza additions and A2A config.

    Args:
        agent_def: Agent definition with optional ``techniques`` and
            ``a2a_skills`` fields.
        defaults: Global defaults with technique declarations.
        workspace: Workspace path for manifest output.

    Returns:
        Dict with keys:
            ``cadenzas``: dict[int, list[dict]] — per-sheet cadenza additions
            ``agent_card``: dict | None — A2A agent card if a2a_skills defined
            ``technique_manifests``: dict[int, str] — per-sheet manifest text
    """
    # Merge default techniques with agent-specific overrides
    merged_techniques = dict(defaults.get("techniques", {}))
    agent_techniques = agent_def.get("techniques", {})
    if isinstance(agent_techniques, dict):
        merged_techniques.update(agent_techniques)

    cadenzas: dict[int, list[dict[str, str]]] = {}
    manifests: dict[int, str] = {}

    # Generate per-sheet technique manifests
    for sheet_num in range(1, SHEETS_PER_CYCLE + 1):
        phase = SHEET_PHASE.get(sheet_num, "")
        active_techniques = self._get_active_techniques(
            merged_techniques, phase
        )
        if active_techniques:
            manifest = self._generate_manifest(active_techniques, phase)
            manifests[sheet_num] = manifest

    # Wire technique document cadenzas
    for tech_name, tech_config in merged_techniques.items():
        if not isinstance(tech_config, dict):
            continue
        kind = tech_config.get("kind", "skill")
        phases = tech_config.get("phases", [])

        # Find technique document if available
        tech_doc_path = self._find_technique_doc(tech_name)
        if tech_doc_path:
            for sheet_num in range(1, SHEETS_PER_CYCLE + 1):
                phase = SHEET_PHASE.get(sheet_num, "")
                if phase in phases or "all" in phases:
                    if sheet_num not in cadenzas:
                        cadenzas[sheet_num] = []
                    cadenzas[sheet_num].append({
                        "file": str(tech_doc_path),
                        "as": "skill" if kind == "skill" else "tool",
                    })

    # Build A2A agent card
    agent_card = self._build_agent_card(agent_def)

    return {
        "cadenzas": cadenzas,
        "agent_card": agent_card,
        "technique_manifests": manifests,
    }

ValidationGenerator

Generates per-sheet validation rules for agent scores.

Produces structural validations (file exists, content checks), CLI instrument validations (temperature/maturity/budget checks), and allows injection of custom validations from the compiler config.

Functions

generate

generate(agent_def, defaults, *, agents_dir='', instruments_dir='')

Generate validation rules for an agent score.

Parameters:

Name	Type	Description	Default
agent_def	dict[str, Any]	Agent definition dict.	required
defaults	dict[str, Any]	Global defaults from compiler config.	required
agents_dir	str	Path to agents identity directory.	''
instruments_dir	str	Path to shared instruments directory.	''

Returns:

Type	Description
list[dict[str, Any]]	List of validation rule dicts for the score YAML.

Source code in src/marianne/compose/validations.py

def generate(
    self,
    agent_def: dict[str, Any],
    defaults: dict[str, Any],
    *,
    agents_dir: str = "",
    instruments_dir: str = "",
) -> list[dict[str, Any]]:
    """Generate validation rules for an agent score.

    Args:
        agent_def: Agent definition dict.
        defaults: Global defaults from compiler config.
        agents_dir: Path to agents identity directory.
        instruments_dir: Path to shared instruments directory.

    Returns:
        List of validation rule dicts for the score YAML.
    """
    name = agent_def["name"]
    validations: list[dict[str, Any]] = []

    # Recon report
    validations.append({
        "type": "file_exists",
        "path": f"{{workspace}}/cycle-state/{name}-recon.md",
        "condition": "stage == 1",
        "description": f"Recon report for {name}",
    })

    # Plan document
    validations.append({
        "type": "file_exists",
        "path": f"{{workspace}}/cycle-state/{name}-plan.md",
        "condition": "stage == 2",
        "description": f"Cycle plan for {name}",
    })

    # User-defined work validations (TDD — test commands etc.)
    for val in defaults.get("validations", []):
        if isinstance(val, dict):
            validations.append({
                "type": "command_succeeds",
                "command": val["command"],
                "condition": "stage == 3",
                "description": val.get("description", val["command"]),
                "timeout_seconds": val.get("timeout_seconds", 600),
            })

    # Temperature check (CLI instrument, sheet 4)
    if instruments_dir and agents_dir:
        validations.append({
            "type": "command_succeeds",
            "command": (
                f"AGENT_DIR={agents_dir}/{name} "
                f"bash {instruments_dir}/temperature-check.sh"
            ),
            "condition": "stage == 4",
            "description": f"Temperature check for {name}",
            "timeout_seconds": 30,
        })

    # Inspection report
    validations.append({
        "type": "file_exists",
        "path": f"{{workspace}}/cycle-state/{name}-inspection.md",
        "condition": "stage == 7",
        "description": f"Inspection report for {name}",
    })

    # User-defined coverage validations (applied to inspect sheets)
    for cov_val in defaults.get("coverage_validations", []):
        if isinstance(cov_val, dict):
            validations.append({
                "type": "command_succeeds",
                "command": cov_val["command"],
                "condition": "stage == 7",
                "description": cov_val.get("description", cov_val["command"]),
                "timeout_seconds": cov_val.get("timeout_seconds", 600),
            })

    # AAR has SUSTAIN and IMPROVE sections
    validations.append({
        "type": "content_contains",
        "path": f"{{workspace}}/cycle-state/{name}-aar.md",
        "pattern": "SUSTAIN:",
        "condition": "stage == 8",
        "description": f"AAR has SUSTAIN for {name}",
    })
    validations.append({
        "type": "content_contains",
        "path": f"{{workspace}}/cycle-state/{name}-aar.md",
        "pattern": "IMPROVE:",
        "condition": "stage == 8",
        "description": f"AAR has IMPROVE for {name}",
    })

    # Maturity check (CLI instrument, sheet 11)
    if instruments_dir and agents_dir:
        validations.append({
            "type": "command_succeeds",
            "command": (
                f"AGENT_DIR={agents_dir}/{name} "
                f"REPORT_PATH={{workspace}}/cycle-state/maturity-report.yaml "
                f"bash {instruments_dir}/maturity-check.sh"
            ),
            "condition": "stage == 11",
            "description": f"Maturity check for {name}",
            "timeout_seconds": 30,
        })

    # Token budget check on resurrect (sheet 12)
    if instruments_dir and agents_dir:
        validations.append({
            "type": "command_succeeds",
            "command": (
                f"AGENT_DIR={agents_dir}/{name} "
                f"L1_BUDGET=900 L2_BUDGET=1500 L3_BUDGET=1500 "
                f"bash {instruments_dir}/token-budget-check.sh"
            ),
            "condition": "stage == 12",
            "description": f"Token budget for {name}",
            "timeout_seconds": 10,
        })

    # Custom user-defined validations from agent config
    for custom in agent_def.get("validations", []):
        if isinstance(custom, dict):
            validations.append(custom)

    return validations

generate_structural

generate_structural(agent_name, phase)

Generate structural validations for a specific phase.

Used when building validations for a single phase rather than the full lifecycle. Returns rules appropriate for the phase.

Source code in src/marianne/compose/validations.py

def generate_structural(
    self,
    agent_name: str,
    phase: str,
) -> list[dict[str, Any]]:
    """Generate structural validations for a specific phase.

    Used when building validations for a single phase rather than
    the full lifecycle. Returns rules appropriate for the phase.
    """
    # Find the sheet number(s) for this phase
    sheet_nums: list[int] = []
    for snum in range(1, SHEETS_PER_CYCLE + 1):
        if SHEET_PHASE.get(snum) == phase:
            sheet_nums.append(snum)

    if not sheet_nums:
        return []

    validations: list[dict[str, Any]] = []
    stage = sheet_nums[0]

    if phase == "recon":
        validations.append({
            "type": "file_exists",
            "path": f"{{workspace}}/cycle-state/{agent_name}-recon.md",
            "condition": f"stage == {stage}",
            "description": "Recon report exists",
        })
    elif phase == "plan":
        validations.append({
            "type": "file_exists",
            "path": f"{{workspace}}/cycle-state/{agent_name}-plan.md",
            "condition": f"stage == {stage}",
            "description": "Cycle plan exists",
        })
    elif phase == "inspect":
        validations.append({
            "type": "file_exists",
            "path": f"{{workspace}}/cycle-state/{agent_name}-inspection.md",
            "condition": f"stage == {stage}",
            "description": "Inspection report exists",
        })
    elif phase == "aar":
        validations.append({
            "type": "content_contains",
            "path": f"{{workspace}}/cycle-state/{agent_name}-aar.md",
            "pattern": "SUSTAIN:",
            "condition": f"stage == {stage}",
            "description": "AAR has SUSTAIN",
        })
        validations.append({
            "type": "content_contains",
            "path": f"{{workspace}}/cycle-state/{agent_name}-aar.md",
            "pattern": "IMPROVE:",
            "condition": f"stage == {stage}",
            "description": "AAR has IMPROVE",
        })

    return validations