From 1564dd772e378a7a20fe151632e864d38d75caff Mon Sep 17 00:00:00 2001 From: graphlemon Date: Thu, 13 Aug 2026 17:43:23 +0800 Subject: [PATCH] graphlemon: auto-commit uncommitted changes for 6f49e70d-59a6-4e87-bb30-fb0e1202a664-main-checkout-drift --- src/e2e_pipeline.py | 39 ++++++++++++++++++++++++++++++++++++++- src/test_e2e_pipeline.py | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 10e21c6..7a4c74b 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -815,6 +815,24 @@ def _record_planner_review_event(conn, task_id, attempt, verdict, used_existing_ conn.commit() +def _existing_active_run_id(conn, task_id): + """2026-08-13 (real recurrence on task b243dfe8): finds a task_id's + existing non-terminal work_graph_run row directly against the table + ux_work_graph_run_active actually constrains, NOT via artifact rows + (unlike _existing_doc_artifact_contents/_shared_run_id_for_task) -- + a run can exist with a real graph_manifest but zero promoted artifacts + (e.g. interrupted before any node's content was written), which those + artifact-based lookups are blind to. Only one non-terminal row can ever + exist per task_id (the index enforces it), so this is unambiguous when + it returns non-None.""" + row = conn.execute( + "SELECT run_id FROM work_graph_run WHERE task_id = ? AND status NOT IN ('done', 'failed') " + "ORDER BY rowid DESC LIMIT 1", + (task_id,), + ).fetchone() + return row[0] if row else None + + def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_path, timeout=300): """fresh-run (plan_and_review_documents) and resume-from- reviewing_planner_output (task_execution.py) both drive this same core @@ -852,12 +870,31 @@ def _review_planner_output_with_retries(conn, task_id, goal_text, target_folder_ 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. + + 2026-08-13 follow-up (same day, real recurrence on task b243dfe8 after + the fix above already landed): the fix above only threads `prior_run_id` + across retry ITERATIONS WITHIN one call to this function -- it missed a + second, real source of collision: a task can already have an orphaned + non-terminal `work_graph_run` row from a PRIOR, separately-interrupted + execution (e.g. a Planner call that persisted `graph_manifest` but + crashed/got restarted before any node's content was promoted to an + `artifact` row) *before this function is even called at all*. + `_existing_doc_artifact_contents` only finds a run via real artifact + rows, so it's blind to this case -- attempt 1 still started from + `prior_run_id = None` and collided with that leftover row exactly like + before. Seeding `prior_run_id` from `_existing_active_run_id()` (queries + `work_graph_run` directly, the actual table `ux_work_graph_run_active` + enforces uniqueness against, independent of whether artifacts exist + yet) closes this: since the index allows only one non-terminal row per + task_id, ANY such row IS "the" run for this task by construction + (Phase 2c merges Planner/Coder/Validator into one shared run), so + appending into it is always correct, never a wrong-run guess. """ from src import task_execution # see _existing_doc_artifact_contents's docstring for why deferred feedback_suffix = "" verdict = None - prior_run_id = None + prior_run_id = _existing_active_run_id(conn, task_id) 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: diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 66dbece..539216e 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -702,6 +702,56 @@ class ReviewPlannerOutputWithRetriesTest(PipelineTestBase): ).fetchall() self.assertEqual(len(events), 2) + def test_orphaned_preexisting_run_with_no_artifacts_appends_not_collides(self): + """Real prod recurrence (2026-08-13, task b243dfe8, AFTER the fix + above already landed): a task can already have a non-terminal + work_graph_run row with a real graph_manifest but ZERO artifact + rows -- e.g. an earlier Planner call that persisted its manifest + then crashed/got interrupted before any node's content was + promoted to an artifact. _existing_doc_artifact_contents only + finds a run via artifact rows, so it's blind to this case -- + attempt 1 used to still start from prior_run_id=None and collide + 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.""" + 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)), + ) + self.conn.commit() + + def fake_plan_subprocess_run(*args, **kwargs): + argv = args[0] + if not argv or argv[0] != "claude": + return _real_subprocess_run(*args, **kwargs) + prompt = argv[argv.index("-p") + 1] + self.assertIn( + "orphan_sdd", prompt, + "append mode must inject the orphaned run's existing node_ids into the prompt", + ) + manifest = {"nodes": [{"node_id": "sdd", "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.return_value = {"content": "# SDD -- real"} + mock_verdict.return_value = "VERDICT: PASS\nlooks real" + + result = e2e_pipeline._review_planner_output_with_retries( + self.conn, self.task_id, "goal", self.tmp.name, + ) + + 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, "must append into the orphaned row, not INSERT a second one") + self.assertEqual(run_rows[0][0], "run-orphan") + self.assertEqual(result["run_id"], "run-orphan") + class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase): """2026-08-08 regression (Org Graph unification Phase 1 ยง7): a -- libgit2 0.21.4