run

run

Run command for Marianne CLI.

This module implements the mzt run command which routes scores through a running conductor (mzt start). Direct execution is not supported — a running conductor is required (like docker requires dockerd).

The only exception is --dry-run, which validates and displays the execution plan without executing anything (no conductor needed).

Classes

Functions

run

run(config_file=Argument(..., help='Path to YAML score configuration file', exists=True, readable=True), dry_run=Option(False, '--dry-run', '-n', help='Show what would be executed without running'), start_sheet=Option(None, '--start-sheet', '-s', help='Override starting sheet number'), workspace=Option(None, '--workspace', '-w', help="Override workspace directory. Creates the directory if it doesn't exist. Takes precedence over the workspace defined in the YAML config."), json_output=Option(False, '--json', '-j', help='Output result as JSON for machine parsing'), escalation=Option(False, '--escalation', '-e', help='Enable human-in-the-loop escalation for low-confidence sheets'), self_healing=Option(False, '--self-healing', '-H', help='Enable automatic diagnosis and remediation when retries are exhausted'), yes=Option(False, '--yes', '-y', help='Auto-confirm suggested fixes when using --self-healing'), fresh=Option(False, '--fresh', help='Delete existing state before running, ensuring a fresh start. Use this for self-chaining scores or when you want to re-run a completed score from scratch without resuming from previous state.'))

Run a score from a YAML configuration file.

Source code in src/marianne/cli/commands/run.py

def run(
    config_file: Path = typer.Argument(
        ...,
        help="Path to YAML score configuration file",
        exists=True,
        readable=True,
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        "-n",
        help="Show what would be executed without running",
    ),
    start_sheet: int | None = typer.Option(
        None,
        "--start-sheet",
        "-s",
        help="Override starting sheet number",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Override workspace directory. Creates the directory if it doesn't exist. "
        "Takes precedence over the workspace defined in the YAML config.",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output result as JSON for machine parsing",
    ),
    escalation: bool = typer.Option(
        False,
        "--escalation",
        "-e",
        help="Enable human-in-the-loop escalation for low-confidence sheets",
    ),
    self_healing: bool = typer.Option(
        False,
        "--self-healing",
        "-H",
        help="Enable automatic diagnosis and remediation when retries are exhausted",
    ),
    yes: bool = typer.Option(
        False,
        "--yes",
        "-y",
        help="Auto-confirm suggested fixes when using --self-healing",
    ),
    fresh: bool = typer.Option(
        False,
        "--fresh",
        help="Delete existing state before running, ensuring a fresh start. "
        "Use this for self-chaining scores or when you want to re-run a completed score "
        "from scratch without resuming from previous state.",
    ),
) -> None:
    """Run a score from a YAML configuration file."""
    from marianne.core.config import JobConfig

    try:
        config = JobConfig.from_yaml(config_file)
    except yaml.YAMLError as e:
        if json_output:
            output_json({"error": f"YAML syntax error: {e}"})
        else:
            output_error(
                f"YAML syntax error: {e}",
                hints=[
                    "Check for indentation issues or invalid YAML characters.",
                    f"Validate with: mzt validate {config_file}",
                ],
            )
        raise typer.Exit(1) from None
    except Exception as e:
        if json_output:
            output_json({"error": str(e)})
        else:
            output_error(
                str(e),
                hints=[f"Validate with: mzt validate {config_file}"],
            )
        raise typer.Exit(1) from None

    # Validate start_sheet (must be positive if provided)
    start_sheet = validate_start_sheet(start_sheet)

    # Override workspace from CLI if provided
    if workspace is not None:
        config.workspace = Path(workspace).resolve()

    # In quiet mode, skip the config panel
    if not is_quiet() and not json_output:
        instrument_display = config.instrument or config.backend.type
        console.print(Panel(
            f"[bold]{config.name}[/bold]\n"
            f"{config.description or 'No description'}\n\n"
            f"Instrument: {instrument_display}\n"
            f"Sheets: {config.sheet.total_sheets} "
            f"({config.sheet.size} items each)\n"
            f"Workspace: {config.workspace}",
            title="Score Configuration",
        ))

    # Cost warning — alert users when cost tracking is disabled
    if not is_quiet() and not json_output and not config.cost_limits.enabled:
        console.print(
            "\n[yellow]Note:[/yellow] Cost tracking is disabled for this score. "
            "API calls will not be monitored or limited."
        )
        console.print(
            "  [dim]To enable, add to your score:[/dim] "
            "cost_limits: {enabled: true, max_cost_per_job: 10.00}"
        )

    # Validate flag compatibility
    if escalation:
        _msg = (
            "--escalation requires interactive console prompts which are "
            "not available in daemon mode. Escalation is not currently "
            "supported."
        )
        if json_output:
            output_json({"error": _msg})
        else:
            output_error(
                _msg,
                hints=[
                    "Remove the --escalation flag to run without interactive escalation.",
                    "Escalation requires a human-in-the-loop and is not yet "
                    "supported in daemon mode.",
                ],
            )
        raise typer.Exit(1)

    if dry_run:
        if not json_output:
            console.print("\n[yellow]Dry run - not executing[/yellow]")
            _show_dry_run(config, config_file)
        else:
            output_json({
                "dry_run": True,
                "job_name": config.name,
                "total_sheets": config.sheet.total_sheets,
                "workspace": str(config.workspace),
            })
        return

    # Route through daemon (required)
    routed = asyncio.run(
        _try_daemon_submit(
            config_file, workspace, fresh, self_healing, yes, json_output,
            start_sheet=start_sheet,
        ),
    )
    if routed:
        return

    # Daemon not available or submission failed
    if json_output:
        output_json({
            "error": "Marianne conductor is not running. Start with: mzt start",
        })
    else:
        output_error(
            "Marianne conductor is not running.",
            hints=["Start it with: mzt start"],
        )
    raise typer.Exit(1)