From 00d0d9c676b6bdd6c4e94c42cfe5490f650d275a Mon Sep 17 00:00:00 2001 From: bboonnee Date: Thu, 13 Aug 2026 16:56:51 +0800 Subject: [PATCH] fix: replan-into-same-run instead of a second work_graph_run row on non-PASS review --- src/e2e_pipeline.py | 51 +++++++++++++++++++++++++++++++++++++++++++++++---- src/test_e2e_pipeline.py | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/test_task_execution.py | 2 +- 3 files changed, 137 insertions(+), 5 deletions(-) diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 41b0be6..10e21c6 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -634,7 +634,7 @@ def _planner_ext_for(artifact_type): return _PLANNER_ARTIFACT_FILE_EXT.get(artifact_type, "md") -def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): +def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None): """Architecture §11 criterion 1: a real Planner Work Graph (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via 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): "mechanical"` (e.g. reusing the catalog's suggested `check_script_compiles` for a test_script node) needs it registered, same as before. + + `append_to_run_id` (2026-08-13, ux_work_graph_run_active fix -- see + `_review_planner_output_with_retries`'s own docstring): default None + keeps this a fresh `plan_work_graph()` INSERT, unchanged from before + this parameter existed -- the very first Planner call for a task, when + no work_graph_run row exists for it yet. Passed a real run_id (a + rejected planner_reviewer attempt re-planning into the SAME task's + still-non-terminal run, same convention `implement_code()` already + uses for Coder), this appends the freshly-planned BRD/PRD/SDD/Benchmark + nodes into that existing row instead of INSERTing a second one -- + `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. """ plan_result = work_graph_plan.plan_work_graph( - conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout + conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout, + append_to_run_id=append_to_run_id, ) 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: @@ -687,7 +706,7 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): 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, + _planner_ext_for, run_graph_error, only_node_ids=only_node_ids, ) if run_graph_error is not None: @@ -814,11 +833,31 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ full_replan. Phase 2's node-level loop_back (org-graph-unification-and- universal-self-planning-requirements.md Phase 2) is the precise-retry upgrade path. + + 2026-08-13 (ux_work_graph_run_active fix, real prod IntegrityError -- + two live tasks crashed 'sqlite3.IntegrityError: UNIQUE constraint + failed: work_graph_run.task_id' right after a non-PASS planner_review_ + verdict): a rejected attempt used to call `plan_documents()` with no + run_id, which always does a fresh `plan_work_graph()` INSERT -- fine + before `ux_work_graph_run_active` existed, but that index now allows + only ONE non-terminal work_graph_run row per task_id, and the earlier + (rejected) attempt's row is never marked 'done'/'failed' just because + its content was rejected (Phase 2c: one run spans a task's whole + lifecycle, `work_graph_run.status` is not how "rejected" is recorded -- + `planner_review_verdict` execution_events are). `prior_run_id` carries + whatever run_id the previous attempt used (the fresh-plan branch's own + new run_id, or the resume branch's `sdd_run_id`) into the next + attempt's `plan_documents(..., append_to_run_id=prior_run_id)` call -- + same "append into the still-open run" convention `implement_code()` + already uses for Coder, not a new one invented here. The very first + attempt (no prior run_id yet) is unaffected: `append_to_run_id=None` + still does the original fresh INSERT. """ from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred feedback_suffix = "" verdict = None + prior_run_id = None for attempt in range(1, _PLANNER_REVIEW_MAX_ATTEMPTS + 1): existing = _existing_doc_artifact_contents(conn, target_folder_path, task_id) if attempt == 1 else {} if existing: @@ -839,8 +878,12 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ ).fetchone()[0] plan_result = {"run_id": sdd_run_id, "artifact_ids": artifact_ids} else: - plan_result = plan_documents(conn, task_id, goal_text + feedback_suffix, target_folder_path) + plan_result = plan_documents( + conn, task_id, goal_text + feedback_suffix, target_folder_path, + append_to_run_id=prior_run_id, + ) artifact_ids = plan_result["artifact_ids"] + prior_run_id = plan_result["run_id"] sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"]) brd_content = ( diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index b06b51e..66dbece 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -613,6 +613,95 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase): ).fetchall() self.assertEqual(len(events), e2e_pipeline._PLANNER_REVIEW_MAX_ATTEMPTS) + def test_rejected_first_attempt_replans_into_same_run_not_a_second_row(self): + """Real prod incident (2026-08-13): ux_work_graph_run_active allows + only ONE non-terminal work_graph_run row per task_id, and a rejected + planner_reviewer attempt's row is never marked 'done'/'failed' just + because its content was rejected -- a retry that calls + plan_documents() with no run_id to append into does a fresh INSERT + and hits 'sqlite3.IntegrityError: UNIQUE constraint failed: + work_graph_run.task_id'. This test goes through the REAL + plan_work_graph()/work_graph_run INSERT path (only the `claude` + subprocess boundary and trigger_org_node_structured are mocked, + same pattern as PlanDocumentsTest) so the real UNIQUE index is what + actually catches a regression here, not a mocked-away assumption. + Before the fix: raises IntegrityError on the second attempt. After: + both attempts land in the SAME row, and only the newly-planned node + actually re-executes (the first attempt's node is already 'done' + and must be skipped, not silently re-run).""" + + def fake_plan_subprocess_run(*args, **kwargs): + # patch("src.work_graph_plan.subprocess.run", ...) patches the + # GLOBAL subprocess.run (see _fake_plan_subprocess_run_dispatch's + # own comment above) -- only intercept the "claude" planning + # call, pass everything else (artifact_write's `git add`/`git + # commit`, etc) through to the real subprocess.run. + argv = args[0] + if not argv or argv[0] != "claude": + return _real_subprocess_run(*args, **kwargs) + # work_graph_plan.plan_work_graph() injects a "...ADDITIONAL + # nodes only" marker into the prompt in append mode (see its + # own docstring) -- used here the same way a real re-planning + # LLM call is instructed to behave: pick a DIFFERENT node_id + # from the rejected attempt's, not redeclare it. Real content + # differs per attempt too (a genuine re-plan), which is why + # this can't just reuse the first attempt's node_id even if + # append mode didn't already forbid the collision outright. + prompt = argv[argv.index("-p") + 1] + node_id = "sdd_v2" if "ADDITIONAL nodes only" in prompt else "sdd" + manifest = {"nodes": [{"node_id": node_id, "artifact_type": "sdd", "depends_on": []}], "edges": []} + payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"} + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") + + with patch("src.work_graph_plan.subprocess.run", side_effect=fake_plan_subprocess_run), \ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger_structured, \ + patch.object(org_node_trigger, "trigger_org_node") as mock_verdict: + mock_trigger_structured.side_effect = [ + {"content": "# SDD v1 -- rejected"}, + {"content": "# SDD v2 -- fixed"}, + ] + mock_verdict.side_effect = ["VERDICT: FAIL\nnot structured enough", "VERDICT: PASS\nlooks real now"] + + result = e2e_pipeline._review_planner_output_with_retries( + self.conn, self.task_id, "goal", self.tmp.name, + ) + + # Exactly one non-terminal work_graph_run row for this task_id -- + # the whole point of ux_work_graph_run_active. Before the fix this + # assertion is moot: the second plan_documents() call never even + # completes, it raises IntegrityError first. + run_rows = self.conn.execute( + "SELECT run_id, status FROM work_graph_run WHERE task_id = ?", (self.task_id,) + ).fetchall() + self.assertEqual(len(run_rows), 1) + self.assertEqual(run_rows[0][1], "planning") + shared_run_id = run_rows[0][0] + self.assertEqual(result["run_id"], shared_run_id) + + # Both attempts' nodes exist in the SAME run's work_node_execution, + # and the first attempt's already-'done' node was skipped (not + # re-executed) the second time around -- exactly 2 trigger_org_ + # node_structured calls total, not 3. + node_ids = { + row[0] for row in self.conn.execute( + "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,) + ).fetchall() + } + self.assertEqual(node_ids, {"sdd", "sdd_v2"}) + self.assertEqual(mock_trigger_structured.call_count, 2) + + # The final PASS'd artifact is the SECOND attempt's content. + sdd_content = e2e_pipeline._read_artifact_content( + self.conn, self.tmp.name, result["artifact_ids"]["sdd"] + ) + self.assertEqual(sdd_content, "# SDD v2 -- fixed") + + events = self.conn.execute( + "SELECT event_type FROM execution_event WHERE task_id = ? AND event_type = 'planner_review_verdict'", + (self.task_id,), + ).fetchall() + self.assertEqual(len(events), 2) + class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase): """2026-08-08 regression (Org Graph unification Phase 1 §7): a diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 50aa739..6a2578f 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -295,7 +295,7 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): self._insert_task("task-1", status="planning_brd_prd_sdd", assigned_org_node="planner") # deliberately: no _write_artifact() calls at all. - def fake_plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300): + def fake_plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300, append_to_run_id=None): # simulates what a real plan_documents() call would actually do -- # writes a real, validation_status='passed' sdd artifact, since # org_node_lifecycle._REQUIRED_ARTIFACTS["coder"] checks the real -- libgit2 0.21.4