Skip to content

instruments

instruments

Instrument resolver — produces per-agent per-sheet instrument assignments.

Resolution order: 1. Start with defaults for each phase type (recon, plan, work, etc.) 2. Apply per-agent overrides 3. For each sheet, resolve the primary + full fallback chain 4. Emit per_sheet_instruments and per_sheet_instrument_config in the score YAML 5. Every sheet gets the full instrument catalog as its tail — no dead ends

Classes

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 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
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,
    }