Index

commands

Functions

errors

errors(job_id=Argument(..., help='Score ID to show errors for'), sheet=Option(None, '--sheet', '-b', help='Filter errors by specific sheet number'), error_type=Option(None, '--type', '-t', help='Filter by error type: transient, rate_limit, or permanent'), error_code=Option(None, '--code', '-c', help='Filter by error code (e.g., E001, E101)'), verbose=Option(False, '--verbose', '-V', help='Show full stdout/stderr tails for each error'), workspace=Option(None, '--workspace', '-w', help='Workspace directory to search for score state (debug override)', hidden=True), json_output=Option(False, '--json', '-j', help='Output errors as JSON'))

List all errors for a score with detailed information.

Displays errors grouped by sheet, with color-coding by error type: - Red: Permanent errors (non-retriable, fatal) - Yellow: Transient errors (retriable with backoff) - Blue: Rate limit errors (retriable after wait)

Examples:

mzt errors my-job # Show all errors mzt errors my-job --sheet 3 # Errors for sheet 3 only mzt errors my-job --type transient # Only transient errors mzt errors my-job --code E001 # Only timeout errors mzt errors my-job --verbose # Show stdout/stderr details

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

def errors(
    job_id: str = typer.Argument(..., help="Score ID to show errors for"),
    sheet: int | None = typer.Option(
        None,
        "--sheet",
        "-b",
        help="Filter errors by specific sheet number",
    ),
    error_type: str | None = typer.Option(
        None,
        "--type",
        "-t",
        help="Filter by error type: transient, rate_limit, or permanent",
    ),
    error_code: str | None = typer.Option(
        None,
        "--code",
        "-c",
        help="Filter by error code (e.g., E001, E101)",
    ),
    verbose: bool = typer.Option(
        False,
        "--verbose",
        "-V",
        help="Show full stdout/stderr tails for each error",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Workspace directory to search for score state (debug override)",
        hidden=True,
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output errors as JSON",
    ),
) -> None:
    """List all errors for a score with detailed information.

    Displays errors grouped by sheet, with color-coding by error type:
    - Red: Permanent errors (non-retriable, fatal)
    - Yellow: Transient errors (retriable with backoff)
    - Blue: Rate limit errors (retriable after wait)

    Examples:
        mzt errors my-job                   # Show all errors
        mzt errors my-job --sheet 3         # Errors for sheet 3 only
        mzt errors my-job --type transient  # Only transient errors
        mzt errors my-job --code E001       # Only timeout errors
        mzt errors my-job --verbose         # Show stdout/stderr details
    """
    from ._shared import validate_job_id

    job_id = validate_job_id(job_id)
    asyncio.run(_errors_job(job_id, sheet, error_type, error_code, verbose, workspace, json_output))

history

history(job_id=Argument(..., help='Score ID to show execution history for'), sheet=Option(None, '--sheet', '-b', help='Filter by specific sheet number'), limit=Option(50, '--limit', '-n', help='Maximum number of records to show'), workspace=Option(None, '--workspace', '-w', help='Workspace directory to search for score state (debug override)', hidden=True), json_output=Option(False, '--json', '-j', help='Output history as JSON'))

Show execution history for a score.

Displays a table of past execution attempts from the SQLite state backend, including sheet number, attempt number, exit code, duration, and timestamp.

Requires the SQLite state backend (execution history is not available with the JSON backend).

Examples:

mzt history my-job # Show all history mzt history my-job --sheet 3 # History for sheet 3 only mzt history my-job --limit 100 # Show more records mzt history my-job --json # Machine-readable output

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

def history(
    job_id: str = typer.Argument(..., help="Score ID to show execution history for"),
    sheet: int | None = typer.Option(
        None,
        "--sheet",
        "-b",
        help="Filter by specific sheet number",
    ),
    limit: int = typer.Option(
        50,
        "--limit",
        "-n",
        help="Maximum number of records to show",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Workspace directory to search for score state (debug override)",
        hidden=True,
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output history as JSON",
    ),
) -> None:
    """Show execution history for a score.

    Displays a table of past execution attempts from the SQLite state backend,
    including sheet number, attempt number, exit code, duration, and timestamp.

    Requires the SQLite state backend (execution history is not available with
    the JSON backend).

    Examples:
        mzt history my-job                  # Show all history
        mzt history my-job --sheet 3        # History for sheet 3 only
        mzt history my-job --limit 100      # Show more records
        mzt history my-job --json           # Machine-readable output
    """
    from ._shared import validate_job_id

    job_id = validate_job_id(job_id)
    asyncio.run(_history_job(job_id, sheet, limit, workspace, json_output))

logs

logs(job_id=Argument(None, help='Score ID to filter logs for (optional, shows all if not specified)'), workspace=Option(None, '--workspace', '-w', help='Workspace directory to find logs (debug override)', hidden=True), log_file=Option(None, '--file', '-f', help='Specific log file path (overrides workspace default)'), follow=Option(False, '--follow', '-F', help='Follow the log file for new entries (like tail -f)'), lines=Option(50, '--lines', '-n', help='Number of lines to show (0 for all)'), level=Option(None, '--level', '-l', help='Filter by minimum log level (DEBUG, INFO, WARNING, ERROR)'), json_output=Option(False, '--json', '-j', help='Output raw JSON log entries'))

Show or tail log files for a score.

Displays log entries from Marianne log files. Supports both current log files and compressed rotated logs (.gz).

Examples:

mzt logs # Show recent logs mzt logs my-job # Filter by job ID mzt logs --follow # Follow log file (like tail -f) mzt logs --lines 100 # Show last 100 lines mzt logs --level ERROR # Show only ERROR and above mzt logs --json # Output raw JSON entries

Note

Log files are stored at {workspace}/logs/marianne.log by default. Use --file to specify a different log file path.

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

def logs(
    job_id: str | None = typer.Argument(
        None,
        help="Score ID to filter logs for (optional, shows all if not specified)",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Workspace directory to find logs (debug override)",
        hidden=True,
    ),
    log_file: Path | None = typer.Option(
        None,
        "--file",
        "-f",
        help="Specific log file path (overrides workspace default)",
    ),
    follow: bool = typer.Option(
        False,
        "--follow",
        "-F",
        help="Follow the log file for new entries (like tail -f)",
    ),
    lines: int = typer.Option(
        50,
        "--lines",
        "-n",
        help="Number of lines to show (0 for all)",
    ),
    level: str | None = typer.Option(
        None,
        "--level",
        "-l",
        help="Filter by minimum log level (DEBUG, INFO, WARNING, ERROR)",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output raw JSON log entries",
    ),
) -> None:
    """Show or tail log files for a score.

    Displays log entries from Marianne log files. Supports both current log files
    and compressed rotated logs (.gz).

    Examples:
        mzt logs                         # Show recent logs
        mzt logs my-job                  # Filter by job ID
        mzt logs --follow                # Follow log file (like tail -f)
        mzt logs --lines 100             # Show last 100 lines
        mzt logs --level ERROR           # Show only ERROR and above
        mzt logs --json                  # Output raw JSON entries

    Note:
        Log files are stored at {workspace}/logs/marianne.log by default.
        Use --file to specify a different log file path.
    """
    from ._shared import validate_job_id

    if job_id is not None:
        job_id = validate_job_id(job_id)
    configure_global_logging(console)

    # Determine log file path
    ws = workspace or Path.cwd()
    target_log = log_file or get_default_log_path(ws)

    # Check if log file exists
    if not target_log.exists():
        # Try to find any log files in the workspace
        available_logs = find_log_files(ws, target_log)
        if not available_logs:
            console.print(f"[yellow]No log files found at:[/yellow] {target_log}")
            console.print(
                "\n[dim]Hint: Logs are created when running scores with file logging enabled.\n"
                "Use --log-file or --log-format=both with mzt run to enable file logging.[/dim]"
            )
            raise typer.Exit(1)
        # Use the first available log
        target_log = available_logs[0]

    # Parse log level filter
    min_level = 0
    if level:
        level_upper = level.upper()
        if level_upper not in _LEVEL_ORDER:
            console.print(
                f"[red]Invalid log level:[/red] {level}\n"
                "Valid levels: DEBUG, INFO, WARNING, ERROR, CRITICAL"
            )
            raise typer.Exit(1)
        min_level = _LEVEL_ORDER[level_upper]

    follower = LogFollower(
        log_path=target_log,
        job_id=job_id,
        min_level=min_level,
        json_output=json_output,
    )

    # Show log file info
    if not is_quiet() and not json_output:
        console.print(f"[dim]Log file: {target_log}[/dim]")

    # Either follow or display
    if follow:
        follower.follow()
    else:
        follower.display(num_lines=lines if lines > 0 else None)

init

init(score_name=Argument(None, help="Score name (e.g. 'my-project'). Same as --name."), path=Option('.', '--path', '-p', help='Directory to initialize (default: current directory)'), name=Option('my-score', '--name', '-n', help='Name for the starter score'), force=Option(False, '--force', '-f', help='Overwrite existing files'), json_output=Option(False, '--json', '-j', help='Output result as JSON'))

Scaffold a new Marianne project with a starter score.

Creates a starter score YAML and .marianne/ project directory. Edit the score with your task, then run it with the conductor.

Examples:

mzt init mzt init my-project mzt init --path ./my-project mzt init --name data-pipeline mzt init --force mzt init --json

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

def init(
    score_name: str | None = typer.Argument(
        None,
        help="Score name (e.g. 'my-project'). Same as --name.",
    ),
    path: Path = typer.Option(
        ".",
        "--path",
        "-p",
        help="Directory to initialize (default: current directory)",
    ),
    name: str = typer.Option(
        "my-score",
        "--name",
        "-n",
        help="Name for the starter score",
    ),
    force: bool = typer.Option(
        False,
        "--force",
        "-f",
        help="Overwrite existing files",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output result as JSON",
    ),
) -> None:
    """Scaffold a new Marianne project with a starter score.

    Creates a starter score YAML and .marianne/ project directory.
    Edit the score with your task, then run it with the conductor.

    Examples:
        mzt init
        mzt init my-project
        mzt init --path ./my-project
        mzt init --name data-pipeline
        mzt init --force
        mzt init --json
    """
    # Resolve name: positional arg is convenience shorthand for --name.
    # --name flag takes precedence when both are provided (flag is explicit).
    if score_name is not None and name == "my-score":
        name = score_name

    # Validate name before touching the filesystem
    name_error = _validate_name(name)
    if name_error:
        output_error(
            name_error,
            hints=["Use a simple name like 'my-score' or 'data-pipeline'."],
            json_output=json_output,
        )
        raise typer.Exit(1)

    target = path.resolve()
    score_file = target / f"{name}.yaml"
    marianne_dir = target / ".marianne"

    # Safety: refuse to overwrite without --force
    if not force:
        if score_file.exists():
            output_error(
                f"Score already exists: {score_file}",
                severity="warning",
                hints=["Use --force to overwrite."],
                json_output=json_output,
            )
            raise typer.Exit(1)
        if marianne_dir.exists():
            output_error(
                f"Marianne project already initialized: {marianne_dir}",
                severity="warning",
                hints=["Use --force to reinitialize."],
                json_output=json_output,
            )
            raise typer.Exit(1)

    # Create .marianne/ directory
    marianne_dir.mkdir(parents=True, exist_ok=True)

    # Create workspaces/ directory so `mzt validate` passes immediately
    workspaces_dir = target / "workspaces"
    workspaces_dir.mkdir(parents=True, exist_ok=True)

    # Generate and write starter score
    score_content = _generate_starter_score(name)
    score_file.write_text(score_content)

    _logger.info("init.complete", extra={"target_path": str(target), "score_name": name})

    # Output
    if json_output:
        output_json({
            "success": True,
            "name": name,
            "score_file": str(score_file),
            "target_path": str(target),
        })
    else:
        console.print(
            f"\n[bold green]Marianne project initialized[/bold green] in {target}\n"
        )
        console.print(
            f"  Created: [bold]{name}.yaml[/bold]"
            "        (starter score — edit with your task)"
        )
        console.print(
            "  Created: [bold].marianne/[/bold]"
            "              (project config directory)"
        )
        console.print()
        console.print("[dim]Next steps:[/dim]")
        console.print("  0. [bold]mzt doctor[/bold]              (check your environment)")
        console.print(f"  1. Edit [bold]{name}.yaml[/bold] with your task")
        console.print(
            f"  2. [bold]mzt start[/bold] && "
            f"[bold]mzt run {name}.yaml[/bold]"
        )
        console.print(f"  3. [bold]mzt status {name}[/bold] to watch progress")
        console.print()

modify

modify(job_id=Argument(..., help='Score ID to modify'), config=Option(..., '--config', '-c', help='New configuration file', exists=True, readable=True), resume_flag=Option(False, '--resume', '-r', help='Immediately resume with new config after pausing'), wait=Option(False, '--wait', help='Wait for score to pause before resuming (when --resume)'), timeout=Option(60, '--timeout', '-t', help='Timeout in seconds for pause acknowledgment'), json_output=Option(False, '--json', '-j', help='Output result as JSON'))

Apply a new configuration to a score and optionally resume execution.

This is a convenience command that combines pause + config update. If the score is running, it will be paused first. Use --resume to immediately resume with the new configuration.

Examples:

mzt modify my-job --config updated.yaml mzt modify my-job -c new-config.yaml --resume mzt modify my-job -c updated.yaml -r --wait

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

def modify(
    job_id: str = typer.Argument(..., help="Score ID to modify"),
    config: Path = typer.Option(
        ...,
        "--config",
        "-c",
        help="New configuration file",
        exists=True,
        readable=True,
    ),
    resume_flag: bool = typer.Option(
        False,
        "--resume",
        "-r",
        help="Immediately resume with new config after pausing",
    ),
    wait: bool = typer.Option(
        False,
        "--wait",
        help="Wait for score to pause before resuming (when --resume)",
    ),
    timeout: int = typer.Option(
        60,
        "--timeout",
        "-t",
        help="Timeout in seconds for pause acknowledgment",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output result as JSON",
    ),
) -> None:
    """Apply a new configuration to a score and optionally resume execution.

    This is a convenience command that combines pause + config update.
    If the score is running, it will be paused first.
    Use --resume to immediately resume with the new configuration.

    Examples:
        mzt modify my-job --config updated.yaml
        mzt modify my-job -c new-config.yaml --resume
        mzt modify my-job -c updated.yaml -r --wait
    """
    from ._shared import validate_job_id

    job_id = validate_job_id(job_id)
    asyncio.run(
        _modify_job(job_id, config, resume_flag, wait, timeout, json_output)
    )

clear_rate_limits

clear_rate_limits(instrument=Option(None, '--instrument', '-i', help='Clear rate limit for a specific instrument only (e.g. claude-cli)'), json_output=Option(False, '--json', '-j', help='Output result as JSON'))

Clear stale rate limits on instruments.

When a backend rate limit expires but the conductor still has it cached, sheets may stay blocked unnecessarily. This command clears the cached limit so dispatch resumes immediately.

Clears both the rate limit coordinator (used by the scheduler) and the baton's per-instrument state (used by the dispatch loop).

Examples:

mzt clear-rate-limits # Clear all mzt clear-rate-limits -i claude-cli # Clear one instrument mzt clear-rate-limits --json # JSON output

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

def clear_rate_limits(
    instrument: str | None = typer.Option(
        None,
        "--instrument",
        "-i",
        help="Clear rate limit for a specific instrument only (e.g. claude-cli)",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output result as JSON",
    ),
) -> None:
    """Clear stale rate limits on instruments.

    When a backend rate limit expires but the conductor still has it cached,
    sheets may stay blocked unnecessarily. This command clears the cached
    limit so dispatch resumes immediately.

    Clears both the rate limit coordinator (used by the scheduler) and
    the baton's per-instrument state (used by the dispatch loop).

    Examples:
        mzt clear-rate-limits                    # Clear all
        mzt clear-rate-limits -i claude-cli      # Clear one instrument
        mzt clear-rate-limits --json             # JSON output
    """
    asyncio.run(_clear_rate_limits(instrument=instrument, json_output=json_output))

clear

clear(job=Option(None, '--score', '-j', help='Specific score ID(s) to clear. Can be repeated.'), status_filter=Option(None, '--status', '-s', help='Status(es) to clear: failed, completed, cancelled. Can be repeated. Defaults to all terminal statuses.'), older_than=Option(None, '--older-than', help='Only clear scores older than this many seconds.'), yes=Option(False, '--yes', '-y', help='Skip confirmation prompt.'))

Clear completed, failed, and cancelled scores from the conductor registry.

Removes completed, failed, and/or cancelled scores from the conductor's tracking. Running and queued scores are never cleared.

Examples:

mzt clear # Clear all terminal scores mzt clear --job conductor-fix # Clear a specific score mzt clear --status failed # Clear only failed scores mzt clear --status failed -s cancelled # Clear failed + cancelled mzt clear --older-than 3600 # Terminal scores older than 1h mzt clear -y # Skip confirmation

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

def clear(
    job: list[str] | None = typer.Option(
        None,
        "--score",
        "-j",
        help="Specific score ID(s) to clear. Can be repeated.",
    ),
    status_filter: list[str] | None = typer.Option(
        None,
        "--status",
        "-s",
        help="Status(es) to clear: failed, completed, cancelled. "
        "Can be repeated. Defaults to all terminal statuses.",
    ),
    older_than: float | None = typer.Option(
        None,
        "--older-than",
        help="Only clear scores older than this many seconds.",
    ),
    yes: bool = typer.Option(
        False,
        "--yes",
        "-y",
        help="Skip confirmation prompt.",
    ),
) -> None:
    """Clear completed, failed, and cancelled scores from the conductor registry.

    Removes completed, failed, and/or cancelled scores from the conductor's
    tracking. Running and queued scores are never cleared.

    Examples:
        mzt clear                                 # Clear all terminal scores
        mzt clear --job conductor-fix             # Clear a specific score
        mzt clear --status failed                 # Clear only failed scores
        mzt clear --status failed -s cancelled    # Clear failed + cancelled
        mzt clear --older-than 3600               # Terminal scores older than 1h
        mzt clear -y                              # Skip confirmation
    """
    asyncio.run(_clear_jobs(job, status_filter, older_than, yes))

list_jobs

list_jobs(all_jobs=Option(False, '--all', '-a', help='Show all scores including completed, failed, and cancelled'), status_filter=Option(None, '--status', '-s', help='Filter by score status (queued, running, completed, failed, paused)'), limit=Option(20, '--limit', '-l', help='Maximum number of scores to display'), json_output=Option(False, '--json', help='Output as JSON array for machine parsing'), workspace=Option(None, '--workspace', '-w', help='Reserved for future per-workspace filtering.', hidden=True))

List scores from the conductor.

By default shows only active scores (queued, running, paused). Use --all to include completed, failed, and cancelled scores.

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

def list_jobs(
    all_jobs: bool = typer.Option(
        False,
        "--all",
        "-a",
        help="Show all scores including completed, failed, and cancelled",
    ),
    status_filter: str | None = typer.Option(
        None,
        "--status",
        "-s",
        help="Filter by score status (queued, running, completed, failed, paused)",
    ),
    limit: int = typer.Option(
        20,
        "--limit",
        "-l",
        help="Maximum number of scores to display",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        help="Output as JSON array for machine parsing",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Reserved for future per-workspace filtering.",
        hidden=True,
    ),
) -> None:
    """List scores from the conductor.

    By default shows only active scores (queued, running, paused).
    Use --all to include completed, failed, and cancelled scores.
    """
    asyncio.run(_list_jobs(all_jobs, status_filter, limit, workspace, json_output))