diagnose

diagnose

Diagnostic commands for Marianne CLI.

This module implements commands for inspecting job state and debugging issues: - logs: View and follow log files - errors: List errors for a job with filtering - diagnose: Generate comprehensive diagnostic reports

★ Insight ───────────────────────────────────── 1. Layered debugging approach: The three commands form a debugging hierarchy: logs for real-time streaming, errors for filtered error lists, and diagnose for comprehensive reports. Users typically progress through these as they narrow down issues.

1. Synthetic error records: When older state files lack error_history, the commands synthesize ErrorRecord objects from sheet-level error_message fields. This maintains backward compatibility with pre-history state files.

2. Error type inference: The infer_error_type function categorizes errors into permanent/transient/rate_limit based on error_category strings. This enables appropriate color-coding and helps users understand retry behavior. ─────────────────────────────────────────────────

Attributes

Classes

LogFollower

LogFollower(log_path, job_id=None, min_level=0, json_output=False)

Parse, filter, and display structured log entries.

Extracted from the logs() command closures to enable unit testing of log parsing, filtering, and formatting independently.

Parameters:

Name	Type	Description	Default
log_path	Path	Path to the log file.	required
job_id	str | None	Optional job ID filter (None = show all).	None
min_level	int	Minimum log level (0=DEBUG, 1=INFO, 2=WARNING, 3=ERROR, 4=CRITICAL).	0
json_output	bool	If True, output raw JSON instead of formatted lines.	False

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

def __init__(
    self,
    log_path: Path,
    job_id: str | None = None,
    min_level: int = 0,
    json_output: bool = False,
) -> None:
    self.log_path = log_path
    self.job_id = job_id
    self.min_level = min_level
    self.json_output = json_output

Functions

parse_line

parse_line(line)

Parse a JSON log line, returning None if blank.

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

def parse_line(self, line: str) -> dict[str, Any] | None:
    """Parse a JSON log line, returning None if blank."""
    line = line.strip()
    if not line:
        return None
    try:
        result: dict[str, Any] = json_module.loads(line)
        return result
    except json_module.JSONDecodeError:
        return {"event": line, "_raw": True}

should_include

should_include(entry)

Check if a log entry passes the configured filters.

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

def should_include(self, entry: dict[str, Any]) -> bool:
    """Check if a log entry passes the configured filters."""
    if self.job_id:
        if entry.get("job_id", "") != self.job_id:
            return False
    entry_level = entry.get("level", "INFO").upper()
    entry_level_num = _LEVEL_ORDER.get(entry_level, 1)
    return entry_level_num >= self.min_level

format_entry

format_entry(entry)

Format a log entry for Rich console display.

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

def format_entry(self, entry: dict[str, Any]) -> str:
    """Format a log entry for Rich console display."""
    if self.json_output:
        return json_module.dumps(entry)

    if entry.get("_raw"):
        return str(entry.get("event", ""))

    timestamp = entry.get("timestamp", "")
    level_str = entry.get("level", "INFO").upper()
    event = entry.get("event", "")
    component = entry.get("component", "")
    entry_job_id = entry.get("job_id", "")
    sheet_num = entry.get(SHEET_NUM_KEY)

    level_color = _LEVEL_COLORS.get(level_str, "white")

    parts: list[str] = []
    if timestamp:
        if "T" in timestamp:
            ts_short = timestamp.split("T")[1].split("+")[0].split(".")[0]
            parts.append(f"[dim]{ts_short}[/dim]")
        else:
            parts.append(f"[dim]{timestamp[:19]}[/dim]")

    parts.append(f"[{level_color}]{level_str:7}[/{level_color}]")

    if component:
        parts.append(f"[cyan]{component}[/cyan]")
    if entry_job_id:
        parts.append(f"[magenta]{entry_job_id}[/magenta]")
    if sheet_num is not None:
        parts.append(f"[green]sheet:{sheet_num}[/green]")
    parts.append(event)

    extras = {k: v for k, v in entry.items() if k not in _EXCLUDE_KEYS}
    if extras:
        extras_str = " ".join(f"{k}={v}" for k, v in extras.items())
        parts.append(f"[dim]{extras_str}[/dim]")

    return " ".join(parts)

read_lines

read_lines(num_lines=None)

Read lines from the log file (handles .gz compression).

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

def read_lines(self, num_lines: int | None = None) -> list[str]:
    """Read lines from the log file (handles .gz compression)."""
    is_gzip_file = self.log_path.suffix == ".gz"
    all_lines: list[str] = []

    try:
        if is_gzip_file:
            with gzip.open(self.log_path, "rt", encoding="utf-8") as f:
                all_lines = f.readlines()
        else:
            with open(self.log_path, encoding="utf-8") as f:
                all_lines = f.readlines()
    except OSError as e:
        output_error(
            f"Cannot read log file: {e}",
            hints=["Check that the Marianne log file exists at ~/.marianne/marianne.log"],
        )
        return []

    if num_lines and num_lines > 0:
        return all_lines[-num_lines:]
    return all_lines

display

display(num_lines=None)

Display filtered log entries.

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

def display(self, num_lines: int | None = None) -> None:
    """Display filtered log entries."""
    raw_lines = self.read_lines(num_lines)

    if not raw_lines:
        console.print("[dim]No log entries found.[/dim]")
        return

    displayed = 0
    for line in raw_lines:
        entry = self.parse_line(line)
        if entry and self.should_include(entry):
            console.print(self.format_entry(entry))
            displayed += 1

    if displayed == 0:
        console.print("[dim]No log entries match the specified filters.[/dim]")
        if self.job_id:
            console.print(f"[dim]Score ID filter: {self.job_id}[/dim]")

follow

follow()

Follow log file for new entries (like tail -f).

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

def follow(self) -> None:
    """Follow log file for new entries (like tail -f)."""
    console.print(f"[dim]Following log file: {self.log_path}[/dim]")
    console.print("[dim]Press Ctrl+C to stop[/dim]\n")

    file_handle = None
    try:
        file_handle = open(self.log_path, encoding="utf-8")  # noqa: SIM115
        file_handle.seek(0, 2)

        while True:
            line = file_handle.readline()
            if line:
                entry = self.parse_line(line)
                if entry and self.should_include(entry):
                    console.print(self.format_entry(entry))
            else:
                time.sleep(0.5)
                if not self.log_path.exists():
                    console.print(
                        "[yellow]Log file rotated. Waiting for new file...[/yellow]"
                    )
                    file_handle.close()
                    for _ in range(10):
                        time.sleep(1)
                        if self.log_path.exists():
                            file_handle = open(self.log_path, encoding="utf-8")  # noqa: SIM115
                            break
                    else:
                        console.print(
                            "[yellow]Log file not recreated. Stopping.[/yellow]"
                        )
                        return
    except KeyboardInterrupt:
        console.print("\n[dim]Stopped following logs.[/dim]")
    except OSError as e:
        output_error(
            f"Cannot follow log file: {e}",
            hints=["Check that the Marianne log file exists at ~/.marianne/marianne.log"],
        )
        raise typer.Exit(1) from None
    finally:
        if file_handle:
            try:
                file_handle.close()
            except OSError:
                pass

Functions

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)

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))

diagnose

diagnose(job_id=Argument(..., help='Score ID to diagnose'), workspace=Option(None, '--workspace', '-w', help='Workspace directory (for scores not in conductor registry)'), json_output=Option(False, '--json', '-j', help='Output diagnostic report as JSON'), include_logs=Option(False, '--include-logs', help='Inline the last 50 lines from each sheet/hook log file in the output'), resources=Option(False, '--resources', help='Include resource profile (peak memory, CPU-time, syscalls, anomalies)'))

Generate a comprehensive diagnostic report for a score.

The diagnostic report includes: - Score overview and current status - Preflight warnings from all sheets - Prompt metrics (token counts, line counts) - Execution timeline with timing information - All errors with full context and output tails - Log file locations, sizes, and modification times - (with --include-logs) Inline log content from each log file - (with --resources) Resource profile from profiler data

This command is particularly useful for debugging failed scores or understanding why a score is running slowly.

Examples:

mzt diagnose my-job # Full diagnostic report mzt diagnose my-job --json # Machine-readable output mzt diagnose my-job --include-logs # Include inline log content mzt diagnose my-job --resources # Include resource profile

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

def diagnose(
    job_id: str = typer.Argument(..., help="Score ID to diagnose"),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Workspace directory (for scores not in conductor registry)",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output diagnostic report as JSON",
    ),
    include_logs: bool = typer.Option(
        False,
        "--include-logs",
        help="Inline the last 50 lines from each sheet/hook log file in the output",
    ),
    resources: bool = typer.Option(
        False,
        "--resources",
        help="Include resource profile (peak memory, CPU-time, syscalls, anomalies)",
    ),
) -> None:
    """Generate a comprehensive diagnostic report for a score.

    The diagnostic report includes:
    - Score overview and current status
    - Preflight warnings from all sheets
    - Prompt metrics (token counts, line counts)
    - Execution timeline with timing information
    - All errors with full context and output tails
    - Log file locations, sizes, and modification times
    - (with --include-logs) Inline log content from each log file
    - (with --resources) Resource profile from profiler data

    This command is particularly useful for debugging failed scores
    or understanding why a score is running slowly.

    Examples:
        mzt diagnose my-job                 # Full diagnostic report
        mzt diagnose my-job --json          # Machine-readable output
        mzt diagnose my-job --include-logs  # Include inline log content
        mzt diagnose my-job --resources     # Include resource profile
    """
    from ._shared import validate_job_id

    job_id = validate_job_id(job_id)
    asyncio.run(
        _diagnose_job(
            job_id, workspace, json_output,
            include_logs=include_logs, resources=resources,
        )
    )

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))