"""The ``retry`` command: user-driven recovery of FAILED materials (M4).
``FAILED`` is terminal -- a material that exhausted its retry budget never
re-runs on its own. This command lets a user restart such materials from the
CLI instead of hand-editing ``state.sqlite``: each selected FAILED material is
reset to the live state it was claimable from before it failed (the output
state of its last completed stage), so the next ``matmelt run`` re-claims
and re-runs the stage it died on. No item-dir files are touched (the stage
re-runs and overwrites) and the completions timeline is preserved.
Selection (all combinable, AND-ed; default = every FAILED material):
* ``--stage <name>`` -- only materials that failed at that funnel stage.
* ``--materials`` / ``--materials-file`` -- only the named material ids.
Safety: a preview clusters the matches by (stage, reason) and the command asks
for confirmation before mutating (``--yes`` skips it; ``--dry-run`` previews and
exits without changing anything). ``--json`` emits the plan/result and, being
non-interactive, applies only with ``--yes``.
Exit codes: 0 = done (including a clean decline or an empty match), 1 = no run
state exists yet, 2 = config or usage error.
"""
from __future__ import annotations
from pathlib import Path
from matmelt.commands import STATE_DB_FILENAME, resolve_config_path
from matmelt.config import ConfigError, load_config
from matmelt.core.stages.registry import FUNNEL_ORDER, STAGE_OUTPUT_STATE
from matmelt.core.state import (
MaterialRecord,
MaterialState,
StateError,
StateStore,
)
from matmelt.ui.console import (
confirm_retry,
print_json_payload,
render_config_error,
render_error,
render_note,
render_retry_preview,
render_retry_result,
render_warning,
)
def _resume_state(record: MaterialRecord) -> MaterialState:
"""Work out the live state a failed material should resume at.
Parameters
----------
record
The failed material, whose completions timeline is the evidence.
Returns
-------
MaterialState
The output state of its most-recently-completed stage, so the stage it
failed at is the one re-claimed -- or ``NEW`` when it never completed
any.
Notes
-----
Derived from the material's own timeline rather than from the configured
funnel, so the answer is correct regardless of which optional stages this
config happens to include.
"""
last_stage: str | None = None
last_at = float("-inf")
for stage, completed_at in record.completions.items():
if stage in STAGE_OUTPUT_STATE and completed_at > last_at:
last_stage, last_at = stage, completed_at
return STAGE_OUTPUT_STATE[last_stage] if last_stage is not None else MaterialState.NEW
[docs]
def run_retry(
config_path: str,
*,
materials: str | None = None,
materials_file: str | None = None,
stage: str | None = None,
dry_run: bool = False,
assume_yes: bool = False,
as_json: bool = False,
) -> int:
"""Reset selected failed materials so a later run re-runs them.
Parameters
----------
config_path
The config file naming the workspace.
materials
Comma-separated ids to retry.
materials_file
A file of ids, one per line. Combines with ``materials``.
stage
Retry only materials that failed at this stage.
dry_run
Show the preview and change nothing.
assume_yes
Skip the confirmation prompt.
as_json
Emit a JSON payload. Applies only together with ``assume_yes``, since
a prompt on stdout would corrupt the payload.
Returns
-------
int
``0`` on success, including when nothing matched or the user declined;
``1`` when there is no run state; ``2`` for a config error, an unknown
stage, or an unusable selection.
Notes
-----
Each material resumes at the state its *own* timeline implies, so a retry
re-runs the stage that failed rather than the whole funnel, and the
completions timeline is preserved. The flow is preview, then prompt, then
result -- a user sees what will change before agreeing to it.
"""
config_path = resolve_config_path(config_path)
try:
config = load_config(config_path)
except ConfigError as exc:
render_config_error(str(exc))
return 2
if stage is not None and stage not in FUNNEL_ORDER:
render_error(f"retry: unknown stage {stage!r}; valid stages: {', '.join(FUNNEL_ORDER)}")
return 2
selection = _parse_selection(materials, materials_file)
if isinstance(selection, int): # error path returned an exit code
return selection
db_path = Path(config.run.work_dir) / STATE_DB_FILENAME
if not db_path.is_file():
render_error(f"retry: no run state at {db_path} -- start one with 'matmelt run'.")
return 1
store = StateStore(db_path)
try:
failed = store.query({"state": MaterialState.FAILED})
chosen = [
record
for record in failed
if (stage is None or record.fail_stage == stage)
and (selection is None or record.material_id in selection)
]
if selection is not None:
unmatched = sorted(selection - {record.material_id for record in chosen})
if unmatched:
render_warning(
f"retry: {len(unmatched)} selected id(s) are not FAILED (or unknown), "
f"skipped: {', '.join(unmatched)}"
)
if not chosen:
render_note("retry: no FAILED materials matched -- nothing to do.")
if as_json:
print_json_payload({"run": config.run.name, "matched": [], "reset": 0})
return 0
plan = [(record, _resume_state(record)) for record in chosen]
if as_json: # non-interactive: preview is the payload, apply only with --yes
apply = not dry_run and assume_yes
reset_count = _apply(store, plan) if apply else 0
print_json_payload(
{
"run": config.run.name,
"matched": [
{
"id": record.material_id,
"stage": record.fail_stage,
"reason": record.fail_reason,
"resume_state": target.value,
}
for record, target in plan
],
"applied": apply,
"reset": reset_count,
}
)
return 0
# Human flow: show the preview BEFORE asking to confirm.
render_retry_preview(config.run.name, chosen)
if dry_run:
render_note("retry --dry-run: nothing changed.")
return 0
if not (assume_yes or confirm_retry(len(chosen))):
render_note("retry: aborted -- nothing changed.")
return 0
render_retry_result(_apply(store, plan))
return 0
finally:
store.close()
def _parse_selection(
materials: str | None, materials_file: str | None
) -> frozenset[str] | int | None:
"""Combine the two selection options into one set of ids.
Parameters
----------
materials
Comma-separated ids, or ``None``.
materials_file
A file of ids, or ``None``. Blank lines and ``#`` comments are skipped.
Returns
-------
frozenset or int or None
The union of both sources; ``None`` when neither was given, meaning
"everything"; or the exit code ``2`` when the file cannot be read or an
explicit selection came out empty. The caller propagates that code --
an empty *explicit* selection is a mistake worth refusing, since
silently meaning "all" would retry far more than asked.
"""
if materials is None and materials_file is None:
return None
picked = {part.strip() for part in (materials or "").split(",") if part.strip()}
if materials_file is not None:
try:
lines = Path(materials_file).read_text(encoding="utf-8").splitlines()
except OSError as exc:
render_error(f"retry: cannot read --materials-file {materials_file}: {exc}")
return 2
picked.update(
line.strip() for line in lines if line.strip() and not line.lstrip().startswith("#")
)
if not picked:
render_error(
"retry: --materials/--materials-file given but the selection is empty -- "
"name at least one material id."
)
return 2
return frozenset(picked)
def _apply(store: StateStore, plan: list[tuple[MaterialRecord, MaterialState]]) -> int:
"""Apply the planned resets.
Parameters
----------
store
The state store to write to.
plan
``(record, target_state)`` pairs, as previewed to the user.
Returns
-------
int
How many materials were actually reset. A material that is no longer
``FAILED`` -- a live run settled it between the preview and the
confirmation -- is a skipped warning, never a crash.
"""
reset_count = 0
for record, target in plan:
try:
store.reset(record.material_id, target)
reset_count += 1
except StateError as exc:
render_warning(f"retry: could not reset {record.material_id}: {exc}")
return reset_count