graphlemon: auto-commit uncommitted changes for 6f49e70d-59a6-4e87-bb30-fb0e1202…

…a664-main-checkout-drift

graphlemon: auto-commit uncommitted changes for 6f49e70d-59a6-4e87-bb30-fb0e1202…
…a664-main-checkout-drift
graphlemon
1 parent 181d5244
Showing 2 changed files with 88 additions and 1 deletions
src/e2e_pipeline.py
src/test_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:
@@ -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