fix: node completion never permanently blocked from artifact promotion

Real incident (2026-08-13, task b243dfe8), diagnosed live with the user after they correctly challenged the architecture: "every node should have its own output, and node-to-node communication IS input/output flowing through edges -- why doesn't a node have an artifact?" They were right to push on this. work_node_execution.output_payload (the real Work Graph data-flow layer) was never the problem -- it's populated the instant any node completes, exactly as expected. The actual bug was one level up: _write_completed_node_artifacts's OLD `only_node_ids` parameter skipped any node not in "this specific append call's own newly-planned nodes", assuming "not new to this call" means "an earlier stage already promoted it". That assumption breaks the instant an earlier call's OWN promotion attempt never actually ran (e.g. it crashed even earlier in that same call, before reaching this function) -- which is exactly what happened to b243dfe8 across multiple retries today: brd/prd/sdd/benchmark nodes kept completing successfully ('done', real content), but zero artifact rows and zero .graphlemon/runs/<task_id>/ files were EVER created, going back to the task's original 2026-08-04 run. Every later append-mode retry kept treating those nodes as "not new, must already be handled" and excluded them forever. Renamed to `new_node_ids`, semantics inverted to check the real source of truth: a node whose artifact_type already has a genuine artifact row for this run_id is skipped UNLESS the node itself is newly-planned by this call (which correctly allows a retry-loop's superseding content to still promote a fresh artifact over a rejected one -- the scenario the previous attempt at this fix broke, caught by the existing test_rejected_first_attempt_replans_into_same_run_not_a_second_row test). New tests cover both real scenarios directly, plus a fix to test_orphaned_preexisting_run_with_no_artifacts_appends_not_collides's own setup (the orphaned node needs to be seeded as already-done, matching the real incident, not left for run_graph() to execute for real a second time). 822/822 tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

fix: node completion never permanently blocked from artifact promotion
Real incident (2026-08-13, task b243dfe8), diagnosed live with the user after they correctly challenged the architecture: "every node should have its own output, and node-to-node communication IS input/output flowing through edges -- why doesn't a node have an artifact?" They were right to push on this. work_node_execution.output_payload (the real Work Graph data-flow layer) was never the problem -- it's populated the instant any node completes, exactly as expected. The actual bug was one level up: _write_completed_node_artifacts's OLD `only_node_ids` parameter skipped any node not in "this specific append call's own newly-planned nodes", assuming "not new to this call" means "an earlier stage already promoted it". That assumption breaks the instant an earlier call's OWN promotion attempt never actually ran (e.g. it crashed even earlier in that same call, before reaching this function) -- which is exactly what happened to b243dfe8 across multiple retries today: brd/prd/sdd/benchmark nodes kept completing successfully ('done', real content), but zero artifact rows and zero .graphlemon/runs/<task_id>/ files were EVER created, going back to the task's original 2026-08-04 run. Every later append-mode retry kept treating those nodes as "not new, must already be handled" and excluded them forever. Renamed to `new_node_ids`, semantics inverted to check the real source of truth: a node whose artifact_type already has a genuine artifact row for this run_id is skipped UNLESS the node itself is newly-planned by this call (which correctly allows a retry-loop's superseding content to still promote a fresh artifact over a rejected one -- the scenario the previous attempt at this fix broke, caught by the existing test_rejected_first_attempt_replans_into_same_run_not_a_second_row test). New tests cover both real scenarios directly, plus a fix to test_orphaned_preexisting_run_with_no_artifacts_appends_not_collides's own setup (the orphaned node needs to be seeded as already-done, matching the real incident, not left for run_graph() to execute for real a second time). 822/822 tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 169d1bae
Showing 2 changed files with 125 additions and 27 deletions
src/e2e_pipeline.py
src/test_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/<task_id>/ 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:
@@ -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):