From 5d6a9a68b79ddb8e3e60d0ec6e2dc9883b236ee6 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Thu, 13 Aug 2026 20:56:17 +0800 Subject: [PATCH] fix: node completion never permanently blocked from artifact promotion --- src/e2e_pipeline.py | 74 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------- src/test_e2e_pipeline.py | 78 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------- 2 files changed, 125 insertions(+), 27 deletions(-) diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 7a4c74b..511705f 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -552,7 +552,7 @@ def _resolve_node_artifact_type(node, node_artifact_types): def _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, ext_for, - run_graph_error=None, only_node_ids=None, + run_graph_error=None, new_node_ids=None, ): """Writes an artifact for every node that's genuinely done AND verify_fn-accepted, sourced from `work_node_execution` in the DB -- @@ -580,12 +580,42 @@ def _write_completed_node_artifacts( 2026-08-12 (Phase 2 §2.3/§6): a node with no resolvable artifact_type (see `_resolve_node_artifact_type`) is silently skipped, not an error -- most Coder/Validator self-planned nodes are internal graph - steps, not final deliverables. `only_node_ids` (default None = every - node in `graph_manifest`, unchanged behavior) restricts processing to - a subset -- needed once a caller works over a MERGED graph (Coder's - `append_to_run_id`) and must avoid re-processing/re-promoting nodes an - EARLIER stage already wrote artifacts for. - """ + steps, not final deliverables. + + 2026-08-13 (real incident, task b243dfe8 -- diagnosed live with the + user, correcting an architectural misunderstanding they caught): the + OLD `only_node_ids` param skipped any node not in "this call's own + newly-planned nodes", on the assumption that "not new to this call" + means "an earlier stage already promoted it". That assumption breaks + the instant an earlier append's OWN promotion attempt never actually + ran -- e.g. this exact function being reached only via a code path + that crashed earlier in THAT call, before ever getting here. The + node's real, valid, already-'done' content in work_node_execution. + output_payload then becomes permanently unpromotable: every LATER + retry (in append mode) kept excluding it as "not new", forever, even + though zero artifact of that type had EVER actually been written -- + exactly what happened to this task (multiple 'done' brd/prd/sdd/ + benchmark nodes across retries, zero artifact rows, zero + .graphlemon/runs// files ever created, going back to the + task's original 2026-08-04 run). + + Renamed to `new_node_ids` and its role inverted: a node whose type + ALREADY has a real artifact row for this run_id is skipped UNLESS the + node itself is in `new_node_ids` (default None -- treated as "nothing + is new", the safe default for a caller with no such concept). This + covers both real scenarios correctly, which a pure "already promoted + -> always skip" rule (tried first, broke the retry-loop's own + already-passing regression test -- a rejected attempt's superseding + content MUST be able to promote a fresh artifact of the same type) + can't distinguish on its own: + - an OLD node (not new to this call) whose type was never actually + promoted (this task's real bug): not in already_promoted_types -> + promoted normally, regardless of new_node_ids. + - an OLD node whose type genuinely was already promoted by an earlier, + successful call: skipped -- no duplicate. + - a NEW node (in new_node_ids) whose type happens to already have an + artifact from a REJECTED earlier attempt (the retry-loop case): + promoted anyway, correctly superseding the rejected one.""" done_rows = conn.execute( "SELECT node_id, output_payload FROM work_node_execution WHERE run_id = ? AND status = 'done'", (run_id,), @@ -598,16 +628,23 @@ def _write_completed_node_artifacts( for excluded in _failed_node_ids(run_graph_error): done_outputs.pop(excluded, None) + already_promoted_types = { + row[0] for row in conn.execute( + "SELECT artifact_type FROM artifact WHERE task_id = ? AND produced_by_run_id = ?", (task_id, run_id) + ).fetchall() + } + new_node_ids = new_node_ids if new_node_ids is not None else set() + artifact_ids = {} for node in work_graph_orchestrate.topological_order(graph_manifest): node_id = node["node_id"] if node_id not in done_outputs: continue - if only_node_ids is not None and node_id not in only_node_ids: - continue artifact_type = _resolve_node_artifact_type(node, node_artifact_types) if artifact_type is None: continue + if artifact_type in already_promoted_types and node_id not in new_node_ids: + continue deps = [ artifact_ids[node_artifact_types[dep_id]] for dep_id in (node.get("depends_on") or []) @@ -619,6 +656,7 @@ def _write_completed_node_artifacts( done_outputs[node_id]["content"], depends_on_artifact_ids=deps, ) artifact_ids[artifact_type] = result["artifact_id"] + already_promoted_types.add(artifact_type) return artifact_ids @@ -680,10 +718,13 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, ap `ux_work_graph_run_active` allows only one non-terminal row per task_id, and the earlier attempt's row is never finalized to 'done'/'failed' just because its content was rejected (Phase 2c: one - run spans a task's whole lifecycle). `only_node_ids` is threaded into - `_write_completed_node_artifacts` the same way `implement_code()` - already does, so a re-plan only (re-)promotes the nodes THIS call - just planned, not the earlier attempt's already-processed ones again. + run spans a task's whole lifecycle). `_write_completed_node_artifacts` + itself now decides what to (re-)promote by checking real artifact rows + for this run_id (2026-08-13 fix, see its own docstring) rather than by + which call planned which node, so a re-plan naturally promotes + whatever hasn't been genuinely promoted yet -- including an earlier + attempt's own already-'done' nodes if THEIR promotion never actually + ran. """ plan_result = work_graph_plan.plan_work_graph( conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout, @@ -692,7 +733,6 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, ap run_id = plan_result["run_id"] graph_manifest = plan_result["graph_manifest"] node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]} - only_node_ids = set(plan_result["new_node_ids"]) if append_to_run_id else None run_graph_error = None try: @@ -706,7 +746,7 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, ap artifact_ids = _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, - _planner_ext_for, run_graph_error, only_node_ids=only_node_ids, + _planner_ext_for, run_graph_error, new_node_ids=set(plan_result["new_node_ids"]), ) if run_graph_error is not None: @@ -1060,7 +1100,7 @@ def implement_code( artifact_ids = _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, - _planner_ext_for, run_graph_error, only_node_ids=new_node_ids, + _planner_ext_for, run_graph_error, new_node_ids=new_node_ids, ) if run_graph_error is not None: @@ -1362,7 +1402,7 @@ def validate_implementation_with_own_graph( artifact_ids = _write_completed_node_artifacts( conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, - _planner_ext_for, run_graph_error, only_node_ids=new_node_ids, + _planner_ext_for, run_graph_error, new_node_ids=new_node_ids, ) if run_graph_error is not None: diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 539216e..a02de4b 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -209,9 +209,9 @@ class PipelineTestBase(unittest.TestCase): class WriteCompletedNodeArtifactsTest(PipelineTestBase): """Phase 2 §2.3/§6 (2026-08-12) direct coverage: promotes_to_artifact - resolution and only_node_ids scoping, tested directly rather than only - incidentally through plan_documents()/implement_code() integration - tests.""" + resolution and (2026-08-13) already-promoted-type dedup, tested + directly rather than only incidentally through plan_documents()/ + implement_code() integration tests.""" def setUp(self): super().setUp() @@ -270,11 +270,10 @@ class WriteCompletedNodeArtifactsTest(PipelineTestBase): with open(os.path.join(self.tmp.name, content)) as f: self.assertEqual(f.read(), "# Real report content") - def test_only_node_ids_scopes_to_new_nodes_skipping_earlier_promotable_ones(self): - # a merged graph (Planner's already-artifact'd nodes + Coder's just- - # appended ones) must not re-promote/re-process nodes an EARLIER - # stage already wrote artifacts for, even though they'd otherwise - # be individually promotable. + def test_node_with_type_already_promoted_this_run_is_not_duplicated(self): + # a node whose artifact_type genuinely already has a real artifact + # row for this run_id must not be re-promoted (no duplicate rows/ + # git commits for unchanged content). manifest = { "nodes": [ {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}, @@ -288,11 +287,54 @@ class WriteCompletedNodeArtifactsTest(PipelineTestBase): self._mark_done("sdd", "# SDD content") self._mark_done("impl", "# Report content") node_artifact_types = {"sdd": "sdd", "impl": "test_report"} + e2e_pipeline._write_completed_node_artifacts( + self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md", + ) + count_after_first_call = self.conn.execute( + "SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1' AND artifact_type = 'sdd'" + ).fetchone()[0] + self.assertEqual(count_after_first_call, 1) + + # a second call (e.g. a later append-mode retry) over the SAME + # manifest/run_id must not create a second "sdd" artifact. artifact_ids = e2e_pipeline._write_completed_node_artifacts( self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md", - only_node_ids={"impl"}, ) - self.assertEqual(set(artifact_ids), {"test_report"}) # NOT "sdd" -- outside only_node_ids + self.assertEqual(artifact_ids, {}) + count_after_second_call = self.conn.execute( + "SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1' AND artifact_type = 'sdd'" + ).fetchone()[0] + self.assertEqual(count_after_second_call, 1) + + def test_earlier_done_node_whose_promotion_never_ran_is_still_promotable_later(self): + # Real incident (2026-08-13, task b243dfe8, diagnosed live with the + # user): a node completed ('done', real output_payload) in an + # EARLIER append-mode call, but that call's OWN + # _write_completed_node_artifacts invocation was never reached + # (e.g. it crashed even earlier in that same call, before getting + # here) -- so no artifact was ever actually written for it. A LATER + # call, over the same merged manifest/run_id, must still be able to + # promote it: real content sitting in work_node_execution. + # output_payload must not become permanently unpromotable just + # because it wasn't part of THIS specific call's own newly-planned + # nodes. + manifest = { + "nodes": [ + {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}, + {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]}, + ], + "edges": [], + } + self._mark_done("sdd", "# SDD content") # done long ago; genuinely never promoted + self._mark_done("benchmark", "# Benchmark content") # newly planned by THIS call + node_artifact_types = {"sdd": "sdd", "benchmark": "benchmark"} + + self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1'").fetchone()[0], 0) + + 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), {"sdd", "benchmark"}) class PlanDocumentsTest(PipelineTestBase): @@ -714,12 +756,28 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase): with this leftover row via the exact same IntegrityError. Seeding prior_run_id from _existing_active_run_id() (queries work_graph_run directly) fixes this.""" + # orphan_sdd is marked 'done' up front (matching the REAL b243dfe8 + # scenario: the node genuinely completed in an earlier, crashed + # attempt) -- without this, run_graph() would treat it as never- + # executed and run it for real via the same mocked + # trigger_org_node_structured() as the "sdd" node below, producing + # byte-identical content for two different node_ids of the same + # artifact_type and hitting a real (but different, not what this + # test is about) "nothing to commit" GitCommitFailedError. orphaned_manifest = {"nodes": [{"node_id": "orphan_sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []} self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-orphan', ?, ?, 'planning')", (self.task_id, json.dumps(orphaned_manifest)), ) + work_graph_orchestrate.ensure_placeholder_skill(self.conn, "planner-doc-writing", "planner") + self.conn.execute( + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " + "input_schema_ref, output_schema_ref, input_payload, output_payload, status) " + "VALUES ('node-exec-orphan-sdd', 'run-orphan', 'orphan_sdd', 'planner', 'planner-doc-writing', " + "'in.json', 'out.json', '{}', ?, 'done')", + (json.dumps({"content": "# SDD -- orphaned, never promoted"}),), + ) self.conn.commit() def fake_plan_subprocess_run(*args, **kwargs): -- libgit2 0.21.4