diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py
index 511705f..4cf9797 100644
--- a/src/e2e_pipeline.py
+++ b/src/e2e_pipeline.py
@@ -625,6 +625,30 @@ def _write_completed_node_artifacts(
reflect_id = node_id + "-reflect"
if reflect_id in done_outputs:
done_outputs[node_id] = done_outputs[reflect_id]
+ # 2026-08-15 (real incident, task b243dfe8): a node that went through
+ # loop_back/retry_self gets a fresh "-attemptN" work_node_execution
+ # row each time (work_graph_orchestrate._next_exec_node_id, N counted
+ # from real loop_back_triggered events) -- graph_manifest itself only
+ # ever names the BASE node_id, never any "-attemptN" variant. Without
+ # this remap, the lookup two lines below (topological_order yields base
+ # node_ids) always resolved to done_outputs[base] -- attempt 1's
+ # content, the FIRST one, permanently stale the instant a node is ever
+ # rejected-then-retried-then-accepted. This node judged 'ok' via its own
+ # verify_fn is real content sitting in the DB; it was simply never
+ # findable under the id this function looks it up by. Highest attempt
+ # number = most recently executed = the one the graph actually moved
+ # past (if the LATEST attempt were still rejected, run_graph() would
+ # either have retried again or raised -- caught by the existing
+ # _failed_node_ids() exclusion below either way, which already operates
+ # on base node_ids and is unaffected by this remap).
+ attempts_by_base = {}
+ for node_id in done_outputs:
+ base, sep, suffix = node_id.rpartition("-attempt")
+ if sep and suffix.isdigit() and base in done_outputs:
+ attempts_by_base.setdefault(base, []).append((int(suffix), node_id))
+ for base, entries in attempts_by_base.items():
+ _, latest_id = max(entries)
+ done_outputs[base] = done_outputs[latest_id]
for excluded in _failed_node_ids(run_graph_error):
done_outputs.pop(excluded, None)
diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py
index a02de4b..d27c442 100644
--- a/src/test_e2e_pipeline.py
+++ b/src/test_e2e_pipeline.py
@@ -336,6 +336,58 @@ class WriteCompletedNodeArtifactsTest(PipelineTestBase):
)
self.assertEqual(set(artifact_ids), {"sdd", "benchmark"})
+ def test_retried_node_promotes_its_latest_attempts_content_not_the_first(self):
+ # Real incident (2026-08-15, task b243dfe8): a node that went
+ # through loop_back/retry_self gets a fresh "-attemptN"
+ # work_node_execution row each retry -- graph_manifest only ever
+ # names the base node_id. Without resolving to the highest attempt
+ # number, this function always found attempt 1's REJECTED content
+ # (the only one filed under the bare node_id), never the later
+ # attempt that actually got accepted -- permanently stale the
+ # instant a node was ever rejected-then-retried-then-accepted.
+ manifest = {
+ "nodes": [{"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": []}],
+ "edges": [],
+ }
+ self._mark_done("benchmark", "# REJECTED first attempt")
+ self._mark_done("benchmark-attempt2", "# also rejected")
+ self._mark_done("benchmark-attempt3", "# finally accepted -- this is the real content")
+ node_artifact_types = {"benchmark": "benchmark"}
+
+ artifact_ids = e2e_pipeline._write_completed_node_artifacts(
+ self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
+ )
+ self.assertEqual(set(artifact_ids), {"benchmark"})
+ content = self.conn.execute(
+ "SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_ids["benchmark"],)
+ ).fetchone()[0]
+ with open(os.path.join(self.tmp.name, content)) as f:
+ self.assertEqual(f.read(), "# finally accepted -- this is the real content")
+
+ def test_exhausted_retry_still_excluded_even_with_multiple_attempts(self):
+ # The _failed_node_ids(run_graph_error) exclusion (existing
+ # behavior) must still work correctly after the attempt-number
+ # remap above -- it operates on base node_ids, which the remap
+ # preserves as dict keys (only the VALUE/content changes).
+ manifest = {
+ "nodes": [{"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": []}],
+ "edges": [],
+ }
+ self._mark_done("benchmark", "# rejected")
+ self._mark_done("benchmark-attempt2", "# rejected again, exhausted budget")
+ run_graph_error = work_graph_orchestrate.ConsistencyCheckFailedError(
+ "loop_back from benchmark to benchmark exceeded max_attempts", node_id="benchmark",
+ )
+ node_artifact_types = {"benchmark": "benchmark"}
+
+ artifact_ids = e2e_pipeline._write_completed_node_artifacts(
+ self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
+ run_graph_error=run_graph_error,
+ )
+ self.assertEqual(artifact_ids, {})
+ count = self.conn.execute("SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1'").fetchone()[0]
+ self.assertEqual(count, 0)
+
class PlanDocumentsTest(PipelineTestBase):
def test_produces_five_artifacts_with_correct_lineage(self):