patterns

patterns

Pattern expander — expands named patterns into sheet sequences.

Patterns from the Rosetta corpus (Cathedral Construction, Composting Cascade, Fan-out + Synthesis, etc.) are available as named patterns the compiler can compose into sheet sequences.

This is the extensibility point: a pattern library that the compiler draws from to produce sheets with the right cognitive structure for the task.

Two expansion modes

• Prompt extension — expand() merges per-phase prompt additions into the agent cycle. Lightweight; enriches existing sheet structure.
• Stage expansion — expand_stages() returns a concrete stage sequence with purposes, instrument guidance, and validation shapes. Structural; defines what the sheet arrangement looks like.

Classes

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.

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