From 1402836df0443efe12252306e3b8f05db8e94c0b Mon Sep 17 00:00:00 2001 From: bboonnee Date: Sat, 15 Aug 2026 10:03:22 +0800 Subject: [PATCH] fix: promote a retried node's latest accepted attempt, not its first --- src/e2e_pipeline.py | 24 ++++++++++++++++++++++++ src/test_e2e_pipeline.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 0 deletions(-) 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): -- libgit2 0.21.4