Commit 00d0d9c676b6bdd6c4e94c42cfe5490f650d275a

Authored by bboonnee
1 parent 947ed31d

fix: replan-into-same-run instead of a second work_graph_run row on non-PASS review

Real prod incident (2026-08-13): two live tasks crashed with sqlite3.IntegrityError
UNIQUE constraint failed: work_graph_run.task_id right after a planner_reviewer
FAIL/INCONCLUSIVE verdict. ux_work_graph_run_active (just added this session) allows
only one non-terminal work_graph_run row per task_id, but a rejected review's retry
called plan_documents() with no run_id, which always INSERTs a fresh row -- the
rejected attempt's own row is never marked done/failed just because its content was
rejected (Phase 2c: one run spans a task's whole lifecycle).

plan_documents() gains append_to_run_id, threaded into plan_work_graph()'s existing
append machinery (same convention implement_code() already uses for Coder).
_review_planner_output_with_retries() now carries prior_run_id across retry attempts
so a re-plan appends into the still-open run instead of colliding with it. The first
attempt (no prior run) is unaffected.

New test drives a real FAIL-then-PASS retry through the actual plan_work_graph()
INSERT path and the real ux_work_graph_run_active index -- confirmed it reproduces
the exact production traceback on the pre-fix code and passes post-fix.

801/801 src tests, 127/127 supervisor tests, both independently re-verified.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
src/e2e_pipeline.py
@@ -634,7 +634,7 @@ def _planner_ext_for(artifact_type): @@ -634,7 +634,7 @@ def _planner_ext_for(artifact_type):
634 return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md") 634 return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md")
635 635
636 636
637 -def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): 637 +def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None):
638 """Architecture §11 criterion 1: a real Planner Work Graph 638 """Architecture §11 criterion 1: a real Planner Work Graph
639 (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via 639 (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
640 work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark, 640 work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark,
@@ -667,13 +667,32 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): @@ -667,13 +667,32 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
667 "mechanical"` (e.g. reusing the catalog's suggested 667 "mechanical"` (e.g. reusing the catalog's suggested
668 `check_script_compiles` for a test_script node) needs it registered, 668 `check_script_compiles` for a test_script node) needs it registered,
669 same as before. 669 same as before.
  670 +
  671 + `append_to_run_id` (2026-08-13, ux_work_graph_run_active fix -- see
  672 + `_review_planner_output_with_retries`'s own docstring): default None
  673 + keeps this a fresh `plan_work_graph()` INSERT, unchanged from before
  674 + this parameter existed -- the very first Planner call for a task, when
  675 + no work_graph_run row exists for it yet. Passed a real run_id (a
  676 + rejected planner_reviewer attempt re-planning into the SAME task's
  677 + still-non-terminal run, same convention `implement_code()` already
  678 + uses for Coder), this appends the freshly-planned BRD/PRD/SDD/Benchmark
  679 + nodes into that existing row instead of INSERTing a second one --
  680 + `ux_work_graph_run_active` allows only one non-terminal row per
  681 + task_id, and the earlier attempt's row is never finalized to
  682 + 'done'/'failed' just because its content was rejected (Phase 2c: one
  683 + run spans a task's whole lifecycle). `only_node_ids` is threaded into
  684 + `_write_completed_node_artifacts` the same way `implement_code()`
  685 + already does, so a re-plan only (re-)promotes the nodes THIS call
  686 + just planned, not the earlier attempt's already-processed ones again.
670 """ 687 """
671 plan_result = work_graph_plan.plan_work_graph( 688 plan_result = work_graph_plan.plan_work_graph(
672 - conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout 689 + conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout,
  690 + append_to_run_id=append_to_run_id,
673 ) 691 )
674 run_id = plan_result["run_id"] 692 run_id = plan_result["run_id"]
675 graph_manifest = plan_result["graph_manifest"] 693 graph_manifest = plan_result["graph_manifest"]
676 node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]} 694 node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]}
  695 + only_node_ids = set(plan_result["new_node_ids"]) if append_to_run_id else None
677 696
678 run_graph_error = None 697 run_graph_error = None
679 try: 698 try:
@@ -687,7 +706,7 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): @@ -687,7 +706,7 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
687 706
688 artifact_ids = _write_completed_node_artifacts( 707 artifact_ids = _write_completed_node_artifacts(
689 conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types, 708 conn, target_folder_path, task_id, run_id, graph_manifest, node_artifact_types,
690 - _planner_ext_for, run_graph_error, 709 + _planner_ext_for, run_graph_error, only_node_ids=only_node_ids,
691 ) 710 )
692 711
693 if run_graph_error is not None: 712 if run_graph_error is not None:
@@ -814,11 +833,31 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ @@ -814,11 +833,31 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_
814 full_replan. Phase 2's node-level loop_back (org-graph-unification-and- 833 full_replan. Phase 2's node-level loop_back (org-graph-unification-and-
815 universal-self-planning-requirements.md Phase 2) is the precise-retry 834 universal-self-planning-requirements.md Phase 2) is the precise-retry
816 upgrade path. 835 upgrade path.
  836 +
  837 + 2026-08-13 (ux_work_graph_run_active fix, real prod IntegrityError --
  838 + two live tasks crashed 'sqlite3.IntegrityError: UNIQUE constraint
  839 + failed: work_graph_run.task_id' right after a non-PASS planner_review_
  840 + verdict): a rejected attempt used to call `plan_documents()` with no
  841 + run_id, which always does a fresh `plan_work_graph()` INSERT -- fine
  842 + before `ux_work_graph_run_active` existed, but that index now allows
  843 + only ONE non-terminal work_graph_run row per task_id, and the earlier
  844 + (rejected) attempt's row is never marked 'done'/'failed' just because
  845 + its content was rejected (Phase 2c: one run spans a task's whole
  846 + lifecycle, `work_graph_run.status` is not how "rejected" is recorded --
  847 + `planner_review_verdict` execution_events are). `prior_run_id` carries
  848 + whatever run_id the previous attempt used (the fresh-plan branch's own
  849 + new run_id, or the resume branch's `sdd_run_id`) into the next
  850 + attempt's `plan_documents(..., append_to_run_id=prior_run_id)` call --
  851 + same "append into the still-open run" convention `implement_code()`
  852 + already uses for Coder, not a new one invented here. The very first
  853 + attempt (no prior run_id yet) is unaffected: `append_to_run_id=None`
  854 + still does the original fresh INSERT.
817 """ 855 """
818 from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred 856 from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred
819 857
820 feedback_suffix = "" 858 feedback_suffix = ""
821 verdict = None 859 verdict = None
  860 + prior_run_id = None
822 for attempt in range(1, _PLANNER_REVIEW_MAX_ATTEMPTS + 1): 861 for attempt in range(1, _PLANNER_REVIEW_MAX_ATTEMPTS + 1):
823 existing = _existing_doc_artifact_contents(conn, target_folder_path, task_id) if attempt == 1 else {} 862 existing = _existing_doc_artifact_contents(conn, target_folder_path, task_id) if attempt == 1 else {}
824 if existing: 863 if existing:
@@ -839,8 +878,12 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ @@ -839,8 +878,12 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_
839 ).fetchone()[0] 878 ).fetchone()[0]
840 plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids} 879 plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids}
841 else: 880 else:
842 - plan_result = plan_documents(conn, task_id, goal_text + feedback_suffix, target_folder_path) 881 + plan_result = plan_documents(
  882 + conn, task_id, goal_text + feedback_suffix, target_folder_path,
  883 + append_to_run_id=prior_run_id,
  884 + )
843 artifact_ids = plan_result["artifact_ids"] 885 artifact_ids = plan_result["artifact_ids"]
  886 + prior_run_id = plan_result["run_id"]
844 887
845 sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"]) 888 sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"])
846 brd_content = ( 889 brd_content = (
src/test_e2e_pipeline.py
@@ -613,6 +613,95 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase): @@ -613,6 +613,95 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase):
613 ).fetchall() 613 ).fetchall()
614 self.assertEqual(len(events), e2e_pipeline._PLANNER_REVIEW_MAX_ATTEMPTS) 614 self.assertEqual(len(events), e2e_pipeline._PLANNER_REVIEW_MAX_ATTEMPTS)
615 615
  616 + def test_rejected_first_attempt_replans_into_same_run_not_a_second_row(self):
  617 + """Real prod incident (2026-08-13): ux_work_graph_run_active allows
  618 + only ONE non-terminal work_graph_run row per task_id, and a rejected
  619 + planner_reviewer attempt's row is never marked 'done'/'failed' just
  620 + because its content was rejected -- a retry that calls
  621 + plan_documents() with no run_id to append into does a fresh INSERT
  622 + and hits 'sqlite3.IntegrityError: UNIQUE constraint failed:
  623 + work_graph_run.task_id'. This test goes through the REAL
  624 + plan_work_graph()/work_graph_run INSERT path (only the `claude`
  625 + subprocess boundary and trigger_org_node_structured are mocked,
  626 + same pattern as PlanDocumentsTest) so the real UNIQUE index is what
  627 + actually catches a regression here, not a mocked-away assumption.
  628 + Before the fix: raises IntegrityError on the second attempt. After:
  629 + both attempts land in the SAME row, and only the newly-planned node
  630 + actually re-executes (the first attempt's node is already 'done'
  631 + and must be skipped, not silently re-run)."""
  632 +
  633 + def fake_plan_subprocess_run(*args, **kwargs):
  634 + # patch("src.work_graph_plan.subprocess.run", ...) patches the
  635 + # GLOBAL subprocess.run (see _fake_plan_subprocess_run_dispatch's
  636 + # own comment above) -- only intercept the "claude" planning
  637 + # call, pass everything else (artifact_write's `git add`/`git
  638 + # commit`, etc) through to the real subprocess.run.
  639 + argv = args[0]
  640 + if not argv or argv[0] != "claude":
  641 + return _real_subprocess_run(*args, **kwargs)
  642 + # work_graph_plan.plan_work_graph() injects a "...ADDITIONAL
  643 + # nodes only" marker into the prompt in append mode (see its
  644 + # own docstring) -- used here the same way a real re-planning
  645 + # LLM call is instructed to behave: pick a DIFFERENT node_id
  646 + # from the rejected attempt's, not redeclare it. Real content
  647 + # differs per attempt too (a genuine re-plan), which is why
  648 + # this can't just reuse the first attempt's node_id even if
  649 + # append mode didn't already forbid the collision outright.
  650 + prompt = argv[argv.index("-p") + 1]
  651 + node_id = "sdd_v2" if "ADDITIONAL nodes only" in prompt else "sdd"
  652 + manifest = {"nodes": [{"node_id": node_id, "artifact_type": "sdd", "depends_on": []}], "edges": []}
  653 + payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
  654 + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
  655 +
  656 + with patch("src.work_graph_plan.subprocess.run", side_effect=fake_plan_subprocess_run), \
  657 + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger_structured, \
  658 + patch.object(org_node_trigger, "trigger_org_node") as mock_verdict:
  659 + mock_trigger_structured.side_effect = [
  660 + {"content": "# SDD v1 -- rejected"},
  661 + {"content": "# SDD v2 -- fixed"},
  662 + ]
  663 + mock_verdict.side_effect = ["VERDICT: FAIL\nnot structured enough", "VERDICT: PASS\nlooks real now"]
  664 +
  665 + result = e2e_pipeline._review_planner_output_with_retries(
  666 + self.conn, self.task_id, "goal", self.tmp.name,
  667 + )
  668 +
  669 + # Exactly one non-terminal work_graph_run row for this task_id --
  670 + # the whole point of ux_work_graph_run_active. Before the fix this
  671 + # assertion is moot: the second plan_documents() call never even
  672 + # completes, it raises IntegrityError first.
  673 + run_rows = self.conn.execute(
  674 + "SELECT run_id, status FROM work_graph_run WHERE task_id = ?", (self.task_id,)
  675 + ).fetchall()
  676 + self.assertEqual(len(run_rows), 1)
  677 + self.assertEqual(run_rows[0][1], "planning")
  678 + shared_run_id = run_rows[0][0]
  679 + self.assertEqual(result["run_id"], shared_run_id)
  680 +
  681 + # Both attempts' nodes exist in the SAME run's work_node_execution,
  682 + # and the first attempt's already-'done' node was skipped (not
  683 + # re-executed) the second time around -- exactly 2 trigger_org_
  684 + # node_structured calls total, not 3.
  685 + node_ids = {
  686 + row[0] for row in self.conn.execute(
  687 + "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,)
  688 + ).fetchall()
  689 + }
  690 + self.assertEqual(node_ids, {"sdd", "sdd_v2"})
  691 + self.assertEqual(mock_trigger_structured.call_count, 2)
  692 +
  693 + # The final PASS'd artifact is the SECOND attempt's content.
  694 + sdd_content = e2e_pipeline._read_artifact_content(
  695 + self.conn, self.tmp.name, result["artifact_ids"]["sdd"]
  696 + )
  697 + self.assertEqual(sdd_content, "# SDD v2 -- fixed")
  698 +
  699 + events = self.conn.execute(
  700 + "SELECT event_type FROM execution_event WHERE task_id = ? AND event_type = 'planner_review_verdict'",
  701 + (self.task_id,),
  702 + ).fetchall()
  703 + self.assertEqual(len(events), 2)
  704 +
616 705
617 class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase): 706 class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase):
618 """2026-08-08 regression (Org Graph unification Phase 1 §7): a 707 """2026-08-08 regression (Org Graph unification Phase 1 §7): a
src/test_task_execution.py
@@ -295,7 +295,7 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): @@ -295,7 +295,7 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase):
295 self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner") 295 self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner")
296 # deliberately: no _write_artifact() calls at all. 296 # deliberately: no _write_artifact() calls at all.
297 297
298 - def fake_plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): 298 + def fake_plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None):
299 # simulates what a real plan_documents() call would actually do -- 299 # simulates what a real plan_documents() call would actually do --
300 # writes a real, validation_status='passed' sdd artifact, since 300 # writes a real, validation_status='passed' sdd artifact, since
301 # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] checks the real 301 # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] checks the real