Respawner prototype

Author: Ladme

src/qq_lib/properties/loop.py

@@ -82,7 +82,7 @@ def __init__(
 
         self.start = start
         self.end = end
-        self.current = current or self._get_cycle()
+        self.current = current or self.determine_cycle_from_archive()
 
         if self.start < 0:
             raise QQError(f"Attribute 'loop-start' ({self.start}) cannot be negative.")
@@ -176,7 +176,7 @@ def to_command_line(self) -> list[str]:
             ":".join(mode.to_str() for mode in self.archive_mode),
         ]
 
-    def _get_cycle(self) -> int:
+    def determine_cycle_from_archive(self) -> int:
         """
         Determine the current cycle number based on files in the archive directory.
 

src/qq_lib/qq.py

@@ -15,6 +15,7 @@
 from qq_lib.killall.cli import killall
 from qq_lib.nodes.cli import nodes
 from qq_lib.queues.cli import queues
+from qq_lib.respawn.cli import respawn
 from qq_lib.run.cli import run
 from qq_lib.shebang.cli import shebang
 from qq_lib.stat.cli import stat
@@ -75,3 +76,4 @@ def cli(ctx: click.Context, version: bool):
 cli.add_command(nodes)
 cli.add_command(shebang)
 cli.add_command(wipe)
+cli.add_command(respawn)

src/qq_lib/respawn/__init__.py

@@ -0,0 +1,2 @@
+# Released under MIT License.
+# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab

src/qq_lib/respawn/cli.py

@@ -0,0 +1,95 @@
+# Released under MIT License.
+# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
+
+import sys
+from pathlib import Path
+from typing import NoReturn
+
+import click
+from rich.console import Console
+
+from qq_lib.core.click_format import GNUHelpColorsCommand
+from qq_lib.core.common import get_info_files
+from qq_lib.core.config import CFG
+from qq_lib.core.error import QQError, QQNotSuitableError
+from qq_lib.core.error_handlers import (
+    handle_general_qq_error,
+    handle_not_suitable_error,
+)
+from qq_lib.core.logger import get_logger
+from qq_lib.core.repeater import Repeater
+from qq_lib.info import Informer
+from qq_lib.respawn.respawner import Respawner
+
+logger = get_logger(__name__)
+console = Console()
+
+
+@click.command(
+    short_help="Respawn a failed/killed job.",
+    help=f"""Respawn the specified qq job, or all qq jobs in the current directory.
+
+{click.style("JOB_ID", fg="green")}   The identifier of the job to respawn. Optional.
+
+If JOB_ID is not specified, `{CFG.binary_name} respawn` searches for qq jobs in the current directory.
+
+Respawning resubmits a failed or killed job to the batch system with its original parameters.
+This is useful when a job fails due to a node failure, an unexpected walltime limit, a random crash,
+or various other types of premature termination.""",
+    cls=GNUHelpColorsCommand,
+    help_options_color="bright_blue",
+)
+@click.argument(
+    "job",
+    type=str,
+    metavar=click.style("JOB_ID", fg="green"),
+    required=False,
+    default=None,
+)
+def respawn(job: str | None) -> NoReturn:
+    try:
+        if job:
+            informers = [Informer.from_job_id(job)]
+        else:
+            if not (
+                informers := [
+                    Informer.from_file(info) for info in get_info_files(Path.cwd())
+                ]
+            ):
+                raise QQError("No qq job info file found.")
+
+        repeater = Repeater(informers, respawn_job)
+        repeater.on_exception(QQNotSuitableError, handle_not_suitable_error)
+        repeater.on_exception(QQError, handle_general_qq_error)
+        repeater.run()
+        print()
+        sys.exit(0)
+    # QQErrors should be caught by Repeater
+    except QQError as e:
+        logger.error(e)
+        sys.exit(CFG.exit_codes.default)
+    except Exception as e:
+        logger.critical(e, exc_info=True, stack_info=True)
+        sys.exit(CFG.exit_codes.unexpected_error)
+
+
+def respawn_job(informer: Informer) -> None:
+    """
+    Attempt to respawn a qq job associated with the specified informer.
+
+    Args:
+        informer (Informer): Informer associated with the job.
+
+    Raises:
+        QQNotSuitableError: If the job is not suitable for respawn.
+        QQError: If the job cannot be respawned.
+    """
+    respawner = Respawner.from_informer(informer)
+    respawner.print_info(console)
+
+    # make sure that the job can actually be respawned
+    respawner.ensure_suitable()
+
+    job_id = respawner.respawn()
+
+    logger.info(f"Job '{informer.info.job_id}' successfully respawned as '{job_id}'.")

src/qq_lib/respawn/respawner.py

@@ -0,0 +1,92 @@
+# Released under MIT License.
+# Copyright (c) 2025-2026 Ladislav Bartos and Robert Vacha Lab
+
+from qq_lib.clear import Clearer
+from qq_lib.core.error import QQError, QQNotSuitableError
+from qq_lib.core.logger import get_logger
+from qq_lib.core.operator import Operator
+from qq_lib.properties.depend import Depend
+from qq_lib.properties.loop import LoopInfo
+from qq_lib.properties.states import RealState
+from qq_lib.submit import Submitter
+
+logger = get_logger(__name__)
+
+
+class Respawner(Operator):
+    def ensure_suitable(self) -> None:
+        """
+        Verify that the job is in a state where it can be respawned.
+
+        Raises:
+            QQNotSuitableError: If the job is in any other state than failed or killed.
+        """
+        if self._state not in {RealState.FAILED, RealState.KILLED}:
+            raise QQNotSuitableError(
+                f"Job cannot be respawned. Job is {str(self._state)}."
+            )
+
+    def respawn(self) -> str:
+        informer = self.get_informer()
+        input_dir = self._info_file.parent
+
+        dependencies = self._handle_dependencies(informer.info.depend)
+        if (loop_info := informer.info.loop_info) is not None:
+            self._ensure_archive_consistent(loop_info)
+
+        submitter = Submitter(
+            batch_system=informer.batch_system,
+            queue=informer.info.queue,
+            account=informer.info.account,
+            script=input_dir / informer.info.script_name,
+            job_type=informer.info.job_type,
+            resources=informer.info.resources,
+            loop_info=informer.info.loop_info,
+            exclude=informer.info.excluded_files,
+            include=informer.info.included_files,
+            depend=dependencies,
+            transfer_mode=informer.info.transfer_mode,
+            server=informer.info.server,
+            interpreter=informer.info.interpreter,
+        )
+
+        # clear files from the input directory
+        clearer = Clearer(input_dir)
+        clearer.clear()
+
+        # respawn the job
+        return submitter.submit()
+
+    def _handle_dependencies(self, dependencies: list[Depend]) -> list[Depend]:
+        """
+        Removes jobs from dependencies that are no longer present in the batch system.
+
+        Without removing these jobs, the respawned job would immediately fail.
+        """
+        BatchSystem = self._informer.batch_system
+
+        filtered = []
+        for depend in dependencies:
+            # get jobs that are still present in the batch system
+            valid_jobs = [
+                job.get_id()
+                for job_id in depend.jobs
+                if not (job := BatchSystem.get_batch_job(job_id)).is_empty()
+            ]
+            if valid_jobs:
+                filtered.append(Depend(depend.type, valid_jobs))
+
+        logger.debug(f"Filtered dependencies: {filtered}.")
+        return filtered
+
+    def _ensure_archive_consistent(self, loop_info: LoopInfo) -> None:
+        """
+        Ensure that the current loop job cycle matches what we would expect based on the contents of the archive directory.
+        """
+        if (
+            archive_cycle := loop_info.determine_cycle_from_archive()
+        ) != loop_info.current:
+            raise QQError(
+                f"Respawning loop job in cycle '{loop_info.current}' but the loop job should continue from cycle '{archive_cycle}' "
+                "based on the contents of the archive directory. Canceling job respawn."
+            )

tests/test_properties_loop.py

@@ -162,72 +162,72 @@ def temp_dir():
 
 def test_get_cycle_returns_start_if_archive_does_not_exist(tmp_path):
     loop_info = _create_loop_info_stub(5, tmp_path / "nonexistent", "md%04d")
-    assert loop_info._get_cycle() == 5
+    assert loop_info.determine_cycle_from_archive() == 5
 
 
 def test_get_cycle_returns_start_if_no_matching_files(temp_dir):
     (temp_dir / "foo.txt").write_text("dummy")
     loop_info = _create_loop_info_stub(2, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 2
+    assert loop_info.determine_cycle_from_archive() == 2
 
 
 def test_get_cycle_selects_highest_number(temp_dir):
     (temp_dir / "md0001.xtc").write_text("x")
     (temp_dir / "md0002.csv").write_text("x")
     (temp_dir / "md0007.txt").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 7
+    assert loop_info.determine_cycle_from_archive() == 7
 
 
 def test_get_cycle_selects_highest_number_partial_match(temp_dir):
     (temp_dir / "md0001.xtc").write_text("x")
     (temp_dir / "md0002.csv").write_text("x")
     (temp_dir / "md0007_px.txt").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 7
+    assert loop_info.determine_cycle_from_archive() == 7
 
 
 def test_get_cycle_selects_highest_number_partial_match2(temp_dir):
     (temp_dir / "md0001.xtc").write_text("x")
     (temp_dir / "md0002.csv").write_text("x")
     (temp_dir / "file_md0007.txt").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 7
+    assert loop_info.determine_cycle_from_archive() == 7
 
 
 def test_get_cycle_files_without_digits_are_ignored(temp_dir):
     (temp_dir / "mdabcd.md").write_text("x")
     (temp_dir / "mdxxxx.txt").write_text("x")
     loop_info = _create_loop_info_stub(3, temp_dir, "md.*")
     # no numerical values in filenames; use start cycle
-    assert loop_info._get_cycle() == 3
+    assert loop_info.determine_cycle_from_archive() == 3
 
 
 def test_get_cycle_mixed_files_some_match_some_not(temp_dir):
     (temp_dir / "md0002.gro").write_text("x")
     (temp_dir / "md25.xtc").write_text("x")  # wrong stem
     (temp_dir / "md0005.mdp").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 5
+    assert loop_info.determine_cycle_from_archive() == 5
 
 
 def test_get_cycle_multiple_digit_sequences_in_stem(temp_dir):
     (temp_dir / "md0003extra123.tpr").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md.*")
-    assert loop_info._get_cycle() == 3
+    assert loop_info.determine_cycle_from_archive() == 3
 
 
 def test_get_cycle_start_value_is_used_as_lower_bound(temp_dir):
     (temp_dir / "md0001.xtc").write_text("x")
     loop_info = _create_loop_info_stub(5, temp_dir, "md%04d")
-    assert loop_info._get_cycle() == 5
+    assert loop_info.determine_cycle_from_archive() == 5
 
 
 def test_get_cycle_non_numeric_files_are_ignored_but_numeric_stems_count(temp_dir):
     (temp_dir / "md0010.xtc").write_text("x")
     (temp_dir / "mdxxxx.txt").write_text("x")
     loop_info = _create_loop_info_stub(0, temp_dir, "md.*")
-    assert loop_info._get_cycle() == 10
+    assert loop_info.determine_cycle_from_archive() == 10
 
 
 def test_to_command_line_basic():