Skip to content

status

status

Status commands for Marianne CLI.

This module implements job status display commands: - mzt status <job-id> - Show detailed status for a specific job - mzt list (list_jobs) - List all jobs in the workspace

★ Insight ───────────────────────────────────── 1. Dual output modes: Both JSON and Rich output are supported for all status commands. JSON mode enables scripting and CI/CD integration while Rich mode provides a human-friendly interactive experience.

  1. Watch mode implementation: The status --watch flag uses a polling loop with screen clearing. This is simpler than event-driven updates but works universally across state backends (JSON files or SQLite).

  2. Circuit breaker inference: The actual CircuitBreaker is a runtime object not persisted to state. We infer its likely state from failure patterns in the persisted sheet states to give operators visibility into retry behavior. ─────────────────────────────────────────────────

Attributes

Classes

CircuitBreakerInference

Bases: TypedDict

Inferred circuit breaker state from job failure patterns.

Functions

format_instrument_with_fallback 

format_instrument_with_fallback(sheet)

Format instrument name with fallback indicator if applicable.

Shows the current instrument and, when fallback history exists, annotates with the previous instrument and reason.

Examples:

"claude-code" (no fallback) "gemini-cli dim[/dim]"

Source code in src/marianne/cli/commands/status.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def format_instrument_with_fallback(sheet: SheetState) -> str:
    """Format instrument name with fallback indicator if applicable.

    Shows the current instrument and, when fallback history exists,
    annotates with the previous instrument and reason.

    Examples:
        "claude-code"  (no fallback)
        "gemini-cli [dim](was claude-code: rate_limit_exhausted)[/dim]"
    """
    name = sheet.instrument_name or ""
    if not sheet.instrument_fallback_history:
        return name
    last = sheet.instrument_fallback_history[-1]
    original = last.get("from", "")
    reason = last.get("reason", "fallback")
    return f"{name} [dim](was {original}: {reason})[/dim]"

status 

status(job_id=Argument(None, help='Score ID to check status for. Omit to see all active scores.'), json_output=Option(False, '--json', '-j', help='Output status as JSON for machine parsing'), watch=Option(False, '--watch', '-W', help='Continuously monitor status with live updates'), watch_interval=Option(5, '--interval', '-i', help='Refresh interval in seconds for --watch mode (default: 5)'), workspace=Option(None, '--workspace', '-w', help='Workspace directory to search for score state (debug override)', hidden=True))

Show score status. Run with no arguments for an overview of all scores.

With no arguments: shows conductor status and all active scores. With a score ID: shows detailed status for that specific score.

Examples:

mzt status mzt status my-job mzt status my-job --json mzt status my-job --watch

Source code in src/marianne/cli/commands/status.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def status(
    job_id: str | None = typer.Argument(
        None,
        help="Score ID to check status for. Omit to see all active scores.",
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        "-j",
        help="Output status as JSON for machine parsing",
    ),
    watch: bool = typer.Option(
        False,
        "--watch",
        "-W",
        help="Continuously monitor status with live updates",
    ),
    watch_interval: int = typer.Option(
        5,
        "--interval",
        "-i",
        help="Refresh interval in seconds for --watch mode (default: 5)",
    ),
    workspace: Path | None = typer.Option(
        None,
        "--workspace",
        "-w",
        help="Workspace directory to search for score state (debug override)",
        hidden=True,
    ),
) -> None:
    """Show score status. Run with no arguments for an overview of all scores.

    With no arguments: shows conductor status and all active scores.
    With a score ID: shows detailed status for that specific score.

    Examples:
        mzt status
        mzt status my-job
        mzt status my-job --json
        mzt status my-job --watch
    """
    if job_id is None:
        asyncio.run(_status_overview(json_output))
        return

    from ._shared import validate_job_id

    job_id = validate_job_id(job_id)
    if watch:
        asyncio.run(_status_job_watch(job_id, json_output, watch_interval, workspace))
    else:
        asyncio.run(_status_job(job_id, json_output, workspace))

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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
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))

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 
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
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))