From 70e4e80aa7d36525b6125d8444e7111519732ad0 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 07:32:15 -0400 Subject: [PATCH 1/5] Resume part way through a loop (phase 5 checkpoints) At the start of each iteration the Loop records its state in the checkpoint (counters, the rows or configurations it was given at entry, the iteration's directory), and passes the next step to step_completed. Resumed, it restores that state, sets the iteration up again without advancing (same directory, iteration.out appended), and starts the body at the step that had not finished; a finished iteration moves on to the next. An iteration that fails and is continued past keeps its writes and is not retried. Tests crash and resume For, Foreach, float, table (whose body appends rows), systems (named directories), nested loops at every position, twice, and with errors continue/exit, comparing the database and directories with an uninterrupted run. Co-Authored-By: Claude Opus 5.5 --- loop_step/loop.py | 133 +++++++++++++-- tests/test_resume.py | 386 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 507 insertions(+), 12 deletions(-) create mode 100644 tests/test_resume.py diff --git a/loop_step/loop.py b/loop_step/loop.py index 7ba40ff..e1ef43a 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -21,6 +21,15 @@ logger = logging.getLogger(__name__) job = printing.getPrinter() + + +def _plain(value): + """A plain Python value for JSON, e.g. from a numpy scalar.""" + if hasattr(value, "item") and not isinstance(value, (list, tuple, dict)): + return value.item() + return value + + printer = printing.getPrinter("loop") @@ -220,6 +229,18 @@ def run(self): # Reset variables to initial state. self._custom_directory_name = None + # Resuming a job part way through this loop? Then the checkpoint holds + # the loop's state and the variables, so the set-up below is skipped and + # the iteration it was in is set up again rather than advanced to. + checkpointer = seamm.checkpoint.get_checkpointer() + resume = None + if checkpointer is not None: + resume = checkpointer.loop_resume(self) + resume_node = None + if resume is not None: + state, resume_node = resume + self._restore_state(state) + # Print out header to the main output printer.important(__(self.description_text(P), indent=self.indent)) @@ -315,6 +336,15 @@ def run(self): else: self.set_variable("_loop_indices", (None,)) elif P["type"] == "For rows in table": + if resume is not None: + self.table = self.get_table(P["table"], create=False) + table_rows = state["items"] + table_indices = state["indices"] + n_table_indices = len(table_indices) + if n_table_indices > 0: + index_is_int = isinstance(table_indices[0], int) + if index_is_int: + fmt = f"0{len(str(max(table_indices) + 1))}d" if self._loop_value is None: self.table = self.get_table(P["table"], create=False) self.table["loop index"] = True @@ -381,7 +411,13 @@ def run(self): elif P["type"] == "For systems in the database": # The configurations to loop over: the standard SEAMM selection. An # empty selection is simply a loop with no iterations. - configurations = self.select_configurations(P, errors=False) + if resume is not None: + system_db = self.get_variable("_system_db") + configurations = [ + system_db.get_configuration(cid) for cid in state["items"] + ] + else: + configurations = self.select_configurations(P, errors=False) if self._loop_value is None: self._loop_value = 0 @@ -425,18 +461,22 @@ def run(self): out_level = out_handler.level out_handler.setLevel(printing.JOB) - # Cycle through the iterations + # Cycle through the iterations. Resuming in the middle of an iteration, + # set it up again without advancing, then start its body at the step + # that had not finished. next_node = self + advance = resume is None or resume_node is None while next_node is not None: if next_node is self: next_node = self.loop_node() if P["type"] == "For": - self._loop_count += 1 - if self._loop_count > 1: - self._loop_value += step - if ndigits > 0: - self._loop_value = round(self._loop_value, ndigits) + if advance: + self._loop_count += 1 + if self._loop_count > 1: + self._loop_value += step + if ndigits > 0: + self._loop_value = round(self._loop_value, ndigits) self.set_variable(P["variable"], self._loop_value) @@ -480,7 +520,8 @@ def run(self): elif P["type"] == "Foreach": self.logger.debug(f"Foreach {P['variable']} in {P['values']}") - self._loop_value += 1 + if advance: + self._loop_value += 1 if self._loop_value > self._loop_length: self._loop_value = None @@ -518,7 +559,8 @@ def run(self): self.set_variable("_loop_index", self._loop_value) self.logger.info(" Loop value = {}".format(value)) elif P["type"] == "For rows in table": - self._loop_value += 1 + if advance: + self._loop_value += 1 if self._loop_value > n_table_indices: self._loop_value = None @@ -578,7 +620,8 @@ def run(self): self.set_variable(key, value) self.logger.debug(" _row = {}".format(row)) elif P["type"] == "For systems in the database": - self._loop_value += 1 + if advance: + self._loop_value += 1 if self._loop_value > self._loop_length: self._loop_value = None @@ -628,6 +671,11 @@ def run(self): self.logger.info(f" system = {system.name}") self.logger.info(f"configuration = {configuration.name}") + # Resuming this iteration: its directory is the one it had, not + # a new unique name next to it. + if not advance: + self._custom_directory_name = state["directory"] + # Direct most output to iteration.out # A handler for the file iter_dir = self.working_path @@ -637,7 +685,8 @@ def run(self): self._file_handler.close() job.removeHandler(self._file_handler) path = iter_dir / "iteration.out" - path.unlink(missing_ok=True) + if advance: + path.unlink(missing_ok=True) self._file_handler = logging.FileHandler(path) self._file_handler.setLevel(printing.NORMAL) formatter = logging.Formatter(fmt="{message:s}", style="{") @@ -650,11 +699,27 @@ def run(self): tmp = self.working_path.name self.set_subids((*self._id, tmp)) + # Checkpoint the start of the iteration, with what is needed to + # set it up again. + if checkpointer is not None: + if not advance: + next_node = self._find_body_node(resume_node) + if P["type"] == "For rows in table": + items = (table_rows, table_indices) + elif P["type"] == "For systems in the database": + items = ([c.id for c in configurations], None) + else: + items = (None, None) + checkpointer.enter_iteration( + self, self._checkpoint_state(P, *items), next_node + ) + advance = True + # Run through the steps in the loop body try: node = next_node next_node = next_node.run() - seamm.step_completed(node) + seamm.step_completed(node, next_node) except DeprecationWarning as e: printer.normal("\nDeprecation warning: " + str(e)) traceback.print_exc(file=sys.stderr) @@ -674,9 +739,15 @@ def run(self): if "continue" in P["errors"]: next_node = self elif "exit" in P["errors"]: + if checkpointer is not None: + checkpointer.iteration_failed(self, node) break else: raise + # Keep what the failed iteration wrote, as before; a resume + # carries on with the next iteration. + if checkpointer is not None: + checkpointer.iteration_failed(self, node) if self.logger.isEnabledFor(logging.DEBUG): p = psutil.Process() @@ -685,6 +756,8 @@ def run(self): self.logger.debug(f"Bottom of loop {next_node}") # Return to the normally scheduled step, i.e. fall out of the loop. + if checkpointer is not None: + checkpointer.leave_loop(self) # Remove any redirection of printing. if self._file_handler is not None: @@ -698,6 +771,42 @@ def run(self): return self.exit_node() + def _checkpoint_state(self, P, items=None, indices=None): + """The loop's state at the start of an iteration, for the checkpoint.""" + if P["type"] == "For": + count = self._loop_count + else: + count = self._loop_value + state = { + "type": P["type"], + "count": count, + "length": self._loop_length, + "loop_count": self._loop_count, + "loop_value": _plain(self._loop_value), + "directory": self._custom_directory_name, + } + if items is not None: + state["items"] = [_plain(x) for x in items] + if indices is not None: + state["indices"] = [_plain(x) for x in indices] + return state + + def _restore_state(self, state): + """Restore the loop's state from the checkpoint, to resume it.""" + self._loop_count = state["loop_count"] + self._loop_value = state["loop_value"] + self._loop_length = state["length"] + self._custom_directory_name = state["directory"] + + def _find_body_node(self, node_id): + """The node of the loop's body with the given id (a list of strings).""" + node = seamm.checkpoint.find_node(self.flowchart, node_id) + if node is not None: + return node + raise seamm.CheckpointError( + f"Cannot find step {'.'.join(node_id)} in the loop to resume at." + ) + def default_edge_subtype(self): """Return the default subtype of the edge. Usually this is 'next' but for nodes with two or more edges leaving them, such as a loop, this diff --git a/tests/test_resume.py b/tests/test_resume.py new file mode 100644 index 0000000..e134f26 --- /dev/null +++ b/tests/test_resume.py @@ -0,0 +1,386 @@ +# -*- coding: utf-8 -*- + +"""Resuming a flowchart part way through a Loop (phase 5 checkpoints). + +Each test runs a flowchart once without interruption, then again in a fresh +directory with a "crash" (an exception no Loop catches, standing in for a kill: +the step's writes are rolled back, as for a killed process) at a chosen step and +iteration, resumes it, and checks that every step ran exactly once overall and +that the job database and directories are the same as the uninterrupted run's. +""" + +import os +import sqlite3 + +import pytest + +import loop_step +import seamm +import seamm_exec.exec_flowchart as ef + + +class Crash(BaseException): + """Stands in for the process being killed.""" + + +@pytest.fixture(autouse=True) +def fresh(monkeypatch): + import seamm_util.argument_parser + + monkeypatch.setattr(seamm_util.argument_parser, "_parsers", {}) + monkeypatch.delenv(ef.RESUME_ENVIRONMENT, raising=False) + Step.runs = [] + Step.crash = None + Step.fail = set() + + +class Step(seamm.Node): + """Creates a system named after itself and the loop indices; can crash.""" + + version = "2026.10.4" + runs = [] + crash = None # (title, indices) to crash at, once + fail = set() # (title, indices) that raise an ordinary error + + def run(self): + indices = self._indices() + key = (self.title, indices) + Step.runs.append(key) + db = self.get_variable("_system_db") + name = f"{self.title}{list(indices)}" + db.create_system(name=name).create_configuration(name=name) + if Step.crash == key or Step.crash == (self.title, None): + Step.crash = None + raise Crash(f"crash at {key}") + if key in Step.fail: + raise RuntimeError(f"failed at {key}") + self.extra(db) + return self.next() + + def extra(self, db): + pass + + def _indices(self): + if self.variable_exists("_loop_indices"): + return tuple(self.get_variable("_loop_indices")) + return () + + +class AppendRow(Step): + """Appends a row to the table being looped over (which must not grow the + loop) and records the current row's name.""" + + def extra(self, db): + table = self.get_table("molecules") + table.set_cell("seen", True) + table.append_row(name=f"added at {self._indices()}", seen=False) + + +class MakeTable(Step): + def extra(self, db): + table = seamm.Table.create( + db, + "molecules", + columns=[("name", "string"), ("seen", "boolean")], + index_column="name", + ) + for name in ("water", "ammonia", "methane", "ethane"): + table.append_row(name=name, seen=False) + self.set_variable("molecules", table) + + +def build(root, body, before=(), after=("After",), **loop_parameters): + """start -> before... -> Join -> Loop(body...) -> after..., as the editor + makes it: the body's last step leads back to the Join.""" + flowchart = seamm.Flowchart(directory=str(root)) + previous = flowchart.get_node("1") + for item in before: + node = item if isinstance(item, seamm.Node) else Step(flowchart, title=item) + flowchart.add_node(node) + flowchart.add_edge(previous, node, edge_type="execution") + previous = node + loop = make_loop(flowchart, **loop_parameters) + add_loop(flowchart, previous, loop, body) + previous = loop + subtype = "exit" + for item in after: + node = item if isinstance(item, seamm.Node) else Step(flowchart, title=item) + flowchart.add_node(node) + flowchart.add_edge(previous, node, edge_type="execution", edge_subtype=subtype) + previous = node + subtype = "next" + flowchart.set_ids() + return flowchart + + +def make_loop(flowchart, **parameters): + loop = loop_step.Loop(flowchart=flowchart) + for key, value in parameters.items(): + loop.parameters[key.replace("_", " ")].value = value + return loop + + +def add_loop(flowchart, previous, loop, body): + """previous -> Join -> Loop, with the body from the Loop back to the Join.""" + join = seamm.Join(flowchart=flowchart) + flowchart.add_node(join) + flowchart.add_node(loop) + flowchart.add_edge(previous, join, edge_type="execution") + flowchart.add_edge(join, loop, edge_type="execution") + attach_body(flowchart, loop, join, body) + + +def attach_body(flowchart, loop, join, body): + previous = loop + subtype = "loop" + for item in body: + if isinstance(item, tuple): # a nested loop: (loop, body) + inner, inner_body = item + add_loop_from(flowchart, previous, subtype, inner, inner_body) + previous = inner + subtype = "exit" + continue + node = item if isinstance(item, seamm.Node) else Step(flowchart, title=item) + flowchart.add_node(node) + flowchart.add_edge(previous, node, edge_type="execution", edge_subtype=subtype) + previous = node + subtype = "next" + flowchart.add_edge(previous, join, edge_type="execution", edge_subtype=subtype) + + +def add_loop_from(flowchart, previous, subtype, loop, body): + join = seamm.Join(flowchart=flowchart) + flowchart.add_node(join) + flowchart.add_node(loop) + flowchart.add_edge(previous, join, edge_type="execution", edge_subtype=subtype) + flowchart.add_edge(join, loop, edge_type="execution") + attach_body(flowchart, loop, join, body) + + +def execute(root, flowchart, resume=False): + options = {"SEAMM": {"resume": resume}} + plan = ef.plan_start(root, options, flowchart, []) + if resume: + assert plan["resume"] is not None, plan["message"] + cwd = os.getcwd() + try: + os.chdir(root) + ef.ExecFlowchart(flowchart, cmdline=[], plan=plan).run(root=str(root)) + finally: + os.chdir(cwd) + + +def database(root): + """What matters in the job database: systems and user tables.""" + db = sqlite3.connect(f"file:{root / 'seamm.db'}?mode=ro", uri=True) + try: + names = [r[0] for r in db.execute("SELECT name FROM system ORDER BY id")] + tables = {} + for (sql_name,) in db.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'table_%'" + ): + tables[sql_name] = db.execute(f"SELECT * FROM {sql_name}").fetchall() + return names, tables + finally: + db.close() + + +def directories(root): + return sorted(str(p.relative_to(root)) for p in root.rglob("*") if p.is_dir()) + + +def crash_and_resume(tmp_path, make, crash): + """Run uninterrupted and crashed+resumed; return both runs' records.""" + straight = tmp_path / "straight" + straight.mkdir() + execute(straight, make(straight)) + expected_runs = list(Step.runs) + + crashed = tmp_path / "crashed" + crashed.mkdir() + Step.runs = [] + Step.crash = crash + flowchart = make(crashed) + with pytest.raises(Crash): + execute(crashed, flowchart) + first = list(Step.runs) + assert first[-1][0] == crash[0] + checkpoint = seamm.read_checkpoint(crashed / "seamm.db") + assert checkpoint["state"] == "error" + + Step.runs = [] + execute(crashed, flowchart, resume=True) + second = list(Step.runs) + # The crashed step runs again; everything before it does not. + assert first[:-1] + second == expected_runs + assert database(crashed) == database(straight) + assert directories(crashed) == directories(straight) + assert seamm.read_checkpoint(crashed / "seamm.db")["state"] == "finished" + return expected_runs, first, second + + +def test_for_loop_crash_in_second_body_step(tmp_path): + def make(root): + return build(root, ["A", "B"], type="For", variable="i", start=1, end=5) + + crash_and_resume(tmp_path, make, ("B", (3,))) + + +def test_for_loop_crash_in_first_body_step(tmp_path): + def make(root): + return build(root, ["A", "B"], type="For", variable="i", start=1, end=5) + + crash_and_resume(tmp_path, make, ("A", (3,))) + + +def test_crash_after_the_loop(tmp_path): + def make(root): + return build(root, ["A"], type="For", variable="i", start=1, end=3) + + expected, first, second = crash_and_resume(tmp_path, make, ("After", ())) + assert second == [("After", ())] + + +def test_foreach(tmp_path): + def make(root): + return build(root, ["A", "B"], type="Foreach", variable="x", values="a b c d") + + crash_and_resume(tmp_path, make, ("B", (2,))) + + +def test_float_for_loop(tmp_path): + def make(root): + return build( + root, ["A"], type="For", variable="T", start=0.1, end=0.5, step=0.1 + ) + + crash_and_resume(tmp_path, make, ("A", (0.3,))) + + +@pytest.mark.parametrize( + "crash", + [ + ("A", (1,)), + ("A", (2,)), + ("B", (2, 1)), + ("C", (2, 2)), + ("C", (2, 3)), + ("D", (2,)), + ("B", (3, 1)), + ("After", ()), + ], +) +def test_nested_loops(tmp_path, crash): + def make(root): + flowchart = seamm.Flowchart(directory=str(root)) + inner = make_loop(flowchart, type="Foreach", variable="x", values="a b c") + outer = make_loop(flowchart, type="For", variable="i", start=1, end=3) + add_loop( + flowchart, flowchart.get_node("1"), outer, ["A", (inner, ["B", "C"]), "D"] + ) + after = Step(flowchart, title="After") + flowchart.add_node(after) + flowchart.add_edge(outer, after, edge_type="execution", edge_subtype="exit") + flowchart.set_ids() + return flowchart + + crash_and_resume(tmp_path, make, crash) + + +def test_table_loop_whose_body_appends_rows(tmp_path): + """The rows appended by the body are not looped over, resumed or not.""" + + def make(root): + flowchart = seamm.Flowchart(directory=str(root)) + return build( + root, + [AppendRow(flowchart, title="Append")], + before=[MakeTable(flowchart, title="Make")], + type="For rows in table", + table="molecules", + ) + + expected, first, second = crash_and_resume(tmp_path, make, ("Append", ("methane",))) + assert len([r for r in expected if r[0] == "Append"]) == 4 + + +def test_errors_continue_then_crash(tmp_path): + """An iteration that failed (and was continued past) is not retried.""" + + def make(root): + return build( + root, + ["A", "B"], + type="For", + variable="i", + start=1, + end=5, + errors="continue to next iteration", + ) + + Step.fail = {("A", (2,))} + expected, first, second = crash_and_resume(tmp_path, make, ("B", (4,))) + assert ("A", (2,)) not in second + + +def test_systems_loop_named_directories(tmp_path): + """Directories named after systems keep their names when resumed.""" + + def make(root): + return build( + root, + ["A", "B"], + before=["S1", "S2", "S3"], + type="For systems in the database", + directory_name="system name", + ) + + crash_and_resume(tmp_path, make, ("B", (2,))) + + +def test_crash_twice(tmp_path): + """A resumed run that crashes again resumes again.""" + + def make(root): + return build(root, ["A", "B"], type="For", variable="i", start=1, end=5) + + straight = tmp_path / "straight" + straight.mkdir() + execute(straight, make(straight)) + expected = list(Step.runs) + + root = tmp_path / "crashed" + root.mkdir() + flowchart = make(root) + runs = [] + for crash, resume in ((("B", (2,)), False), (("A", (4,)), True)): + Step.runs = [] + Step.crash = crash + with pytest.raises(Crash): + execute(root, flowchart, resume=resume) + runs.extend(Step.runs[:-1]) + Step.runs = [] + execute(root, flowchart, resume=True) + runs.extend(Step.runs) + assert runs == expected + assert database(root) == database(straight) + + +def test_errors_exit_the_loop_then_crash(tmp_path): + def make(root): + return build( + root, + ["A", "B"], + after=("After", "Last"), + type="For", + variable="i", + start=1, + end=5, + errors="exit the loop", + ) + + Step.fail = {("B", (3,))} + # (Leaving a loop early leaves its index variables set, as it always has.) + expected, first, second = crash_and_resume(tmp_path, make, ("Last", None)) + assert [r[0] for r in second] == ["Last"] + assert ("A", (4,)) not in expected From 56f2164e8725ed5cec1c7df10a522cd097239af8 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 10:30:42 -0400 Subject: [PATCH 2/5] Resume tests: build the flowchart again before resuming, as a new process does Co-Authored-By: Claude Opus 5.5 --- tests/test_resume.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_resume.py b/tests/test_resume.py index e134f26..27fd464 100644 --- a/tests/test_resume.py +++ b/tests/test_resume.py @@ -209,6 +209,7 @@ def crash_and_resume(tmp_path, make, crash): assert checkpoint["state"] == "error" Step.runs = [] + flowchart = make(crashed) # a resume reads the flowchart again: new uuids execute(crashed, flowchart, resume=True) second = list(Step.runs) # The crashed step runs again; everything before it does not. @@ -351,16 +352,15 @@ def make(root): root = tmp_path / "crashed" root.mkdir() - flowchart = make(root) runs = [] for crash, resume in ((("B", (2,)), False), (("A", (4,)), True)): Step.runs = [] Step.crash = crash with pytest.raises(Crash): - execute(root, flowchart, resume=resume) + execute(root, make(root), resume=resume) runs.extend(Step.runs[:-1]) Step.runs = [] - execute(root, flowchart, resume=True) + execute(root, make(root), resume=True) runs.extend(Step.runs) assert runs == expected assert database(root) == database(straight) From 894c5821893e3f9e31923a6fb7d2e68911fe3c62 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 13:22:18 -0400 Subject: [PATCH 3/5] Release 2026.10.4 prep: loops resume (HISTORY, seamm>=2026.10.4) Co-Authored-By: Claude Opus 5.5 --- HISTORY.rst | 8 ++++++++ requirements.txt | 2 +- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/HISTORY.rst b/HISTORY.rst index 88d72c7..6c21ca5 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -1,6 +1,14 @@ ======= History ======= +2026.10.4 -- Loops resume where they stopped + * A job resumed from its checkpoint (seamm_exec 2026.10.4.1, ``--resume``) + continues a loop at the iteration and step it had reached, with the same + iteration directory and the same rows, structures or values to loop over as + when the loop started; a finished iteration is not run again. An iteration that + failed and was skipped ("continue to next iteration") is not retried. + * Requires seamm 2026.10.4. + 2026.10.3 -- Loops over tables in the job's database * Loops over the rows of a table work with tables kept in the job's database (seamm 2026.10.3); the job's database is committed after each step of the loop body. diff --git a/requirements.txt b/requirements.txt index 80fa3e4..311ea9b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,2 @@ psutil -seamm>=2026.10.3 +seamm>=2026.10.4 From 08851cc4817de556dba9152693ecf436276db8ec Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 13:45:55 -0400 Subject: [PATCH 4/5] Loop: checkpoint an iteration before making its directory A kill between making the directory and writing the checkpoint left a directory the checkpoint did not know, so the resumed iteration got a name_2 directory. A resumed iteration keeps its iteration.out. HISTORY: 'exit the loop' also keeps a failed iteration's writes. Co-Authored-By: Claude Opus 5.5 --- HISTORY.rst | 3 ++- loop_step/loop.py | 38 +++++++++++++++++++------------------ tests/test_resume.py | 45 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 67 insertions(+), 19 deletions(-) diff --git a/HISTORY.rst b/HISTORY.rst index 6c21ca5..bca2dac 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -6,7 +6,8 @@ History continues a loop at the iteration and step it had reached, with the same iteration directory and the same rows, structures or values to loop over as when the loop started; a finished iteration is not run again. An iteration that - failed and was skipped ("continue to next iteration") is not retried. + failed and was skipped ("continue to next iteration" or "exit the loop") + keeps what it wrote, as before, and is not retried. * Requires seamm 2026.10.4. 2026.10.3 -- Loops over tables in the job's database diff --git a/loop_step/loop.py b/loop_step/loop.py index e1ef43a..014beb5 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -676,23 +676,6 @@ def run(self): if not advance: self._custom_directory_name = state["directory"] - # Direct most output to iteration.out - # A handler for the file - iter_dir = self.working_path - iter_dir.mkdir(parents=True, exist_ok=True) - - if self._file_handler is not None: - self._file_handler.close() - job.removeHandler(self._file_handler) - path = iter_dir / "iteration.out" - if advance: - path.unlink(missing_ok=True) - self._file_handler = logging.FileHandler(path) - self._file_handler.setLevel(printing.NORMAL) - formatter = logging.Formatter(fmt="{message:s}", style="{") - self._file_handler.setFormatter(formatter) - job.addHandler(self._file_handler) - # Add the iteration to the ids so the directory structure is # reasonable self.flowchart.reset_visited() @@ -700,7 +683,8 @@ def run(self): self.set_subids((*self._id, tmp)) # Checkpoint the start of the iteration, with what is needed to - # set it up again. + # set it up again -- before its directory exists, so a kill in + # between cannot leave a directory the checkpoint does not know. if checkpointer is not None: if not advance: next_node = self._find_body_node(resume_node) @@ -713,8 +697,26 @@ def run(self): checkpointer.enter_iteration( self, self._checkpoint_state(P, *items), next_node ) + resuming_iteration = not advance advance = True + # Direct most output to iteration.out + # A handler for the file + iter_dir = self.working_path + iter_dir.mkdir(parents=True, exist_ok=True) + + if self._file_handler is not None: + self._file_handler.close() + job.removeHandler(self._file_handler) + path = iter_dir / "iteration.out" + if not resuming_iteration: + path.unlink(missing_ok=True) + self._file_handler = logging.FileHandler(path) + self._file_handler.setLevel(printing.NORMAL) + formatter = logging.Formatter(fmt="{message:s}", style="{") + self._file_handler.setFormatter(formatter) + job.addHandler(self._file_handler) + # Run through the steps in the loop body try: node = next_node diff --git a/tests/test_resume.py b/tests/test_resume.py index 27fd464..5d70954 100644 --- a/tests/test_resume.py +++ b/tests/test_resume.py @@ -384,3 +384,48 @@ def make(root): expected, first, second = crash_and_resume(tmp_path, make, ("Last", None)) assert [r[0] for r in second] == ["Last"] assert ("A", (4,)) not in expected + + +def test_an_error_while_resuming_keeps_the_loop_position(tmp_path, monkeypatch): + """A resume that fails before the Loop re-enters its iteration must leave + the checkpoint where it was, so the next resume continues the loop rather + than starting it again over the committed iterations.""" + + def make(root): + return build(root, ["A", "B"], type="For", variable="i", start=1, end=5) + + straight = tmp_path / "straight" + straight.mkdir() + execute(straight, make(straight)) + expected = list(Step.runs) + + root = tmp_path / "crashed" + root.mkdir() + Step.runs = [] + Step.crash = ("B", (3,)) + with pytest.raises(Crash): + execute(root, make(root)) + runs = list(Step.runs[:-1]) + before = seamm.read_checkpoint(root / "seamm.db")["position"] + + # The resume fails inside the Loop, before it re-enters the iteration + original = loop_step.Loop._restore_state + + def broken(self, state): + raise RuntimeError("cannot restore the loop") + + monkeypatch.setattr(loop_step.Loop, "_restore_state", broken) + Step.runs = [] + with pytest.raises(RuntimeError, match="cannot restore"): + execute(root, make(root), resume=True) + checkpoint = seamm.read_checkpoint(root / "seamm.db") + assert checkpoint["state"] == "error" + assert checkpoint["position"] == before + + monkeypatch.setattr(loop_step.Loop, "_restore_state", original) + Step.runs = [] + execute(root, make(root), resume=True) + runs.extend(Step.runs) + assert runs == expected + assert database(root) == database(straight) + assert directories(root) == directories(straight) From 56fc39b1cdf246bcaa35ced1c32075e75a237090 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 14:37:13 -0400 Subject: [PATCH 5/5] Declare seamm-exec as a test dependency (the resume tests use the evaluator) Co-Authored-By: Claude Opus 5.5 --- setup.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/setup.py b/setup.py index 29d7e8d..99c3187 100644 --- a/setup.py +++ b/setup.py @@ -53,6 +53,9 @@ # deployment install_requires=requirements, + # The resume tests run flowcharts through the evaluator + extras_require={"test": ["seamm-exec>=2026.10.4.1"]}, + test_suite='tests', # Valid platforms your code works on, adjust to your flavor