test: cover the planning-call work_node_execution recorded by _record_planning_node

Adds targeted coverage for the synthetic "__planning__" node work_graph_plan.py now records (real work_node_execution + log, task_query.py display-only prepend) -- the row exists and is done, the persisted work_graph_run.graph_manifest stays pure for execution/fork/append-mode, the log carries the real prompt and manifest, task_query surfaces it as nodes[0] with a clickable node_exec_id, and append mode doesn't duplicate it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

test: cover the planning-call work_node_execution recorded by _record_planning_node
Adds targeted coverage for the synthetic "__planning__" node work_graph_plan.py now records (real work_node_execution + log, task_query.py display-only prepend) -- the row exists and is done, the persisted work_graph_run.graph_manifest stays pure for execution/fork/append-mode, the log carries the real prompt and manifest, task_query surfaces it as nodes[0] with a clickable node_exec_id, and append mode doesn't duplicate it. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 1a8f99c2
Showing 1 changed file with 104 additions and 1 deletions
src/test_work_graph_plan.py
@@ -12,7 +12,7 @@ import subprocess
 import unittest
 from unittest.mock import patch
  
-from src import org_node_trigger, schema, work_graph_orchestrate, work_graph_plan
+from src import execution_log, org_node_trigger, schema, task_query, work_graph_orchestrate, work_graph_plan
  
 _MANIFEST = {
     "nodes": [
@@ -686,5 +686,108 @@ class GenericPathEndToEndViaRunGraphTest(PlanWorkGraphTestBase):
         self.assertEqual(mock_trigger.call_count, 3)
  
  
+class PlanningNodeRecordedTest(PlanWorkGraphTestBase):
+    """The planning call itself (this class's own `run.return_value`
+    subprocess call) must show up as a real, clickable work_node_execution
+    -- not just as the graph_manifest it produced -- so v_task_detail's
+    to-do tree/DAG have a first node for "how was this run planned",
+    per the user's own request (2026-08-15, task b243dfe8 discussion)."""
+
+    def test_planning_node_execution_row_is_real_and_done(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            result = work_graph_plan.plan_work_graph(
+                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
+            )
+        row = self.conn.execute(
+            "SELECT run_id, subagent_type, skill_ref, status, output_payload "
+            "FROM work_node_execution WHERE node_id = ?",
+            (work_graph_plan.PLANNING_NODE_ID,),
+        ).fetchone()
+        self.assertIsNotNone(row)
+        self.assertEqual(row[0], result["run_id"])
+        self.assertEqual(row[1], "planner")
+        self.assertEqual(row[2], "design-dynamic-work-graph")
+        self.assertEqual(row[3], "done")
+        self.assertEqual(json.loads(row[4])["graph_manifest"], _MANIFEST)
+
+    def test_persisted_work_graph_run_manifest_stays_pure_for_execution(self):
+        # run_graph()/fork/append-mode all read work_graph_run.graph_manifest
+        # straight off the DB -- it must never gain the synthetic planning
+        # node, or a later append_to_run_id call / a fork would try to treat
+        # "planning itself" as executable work. See _record_planning_node's
+        # own docstring.
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            result = work_graph_plan.plan_work_graph(
+                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
+            )
+        persisted = self.conn.execute(
+            "SELECT graph_manifest FROM work_graph_run WHERE run_id = ?", (result["run_id"],)
+        ).fetchone()[0]
+        self.assertEqual(json.loads(persisted), _MANIFEST)
+
+    def test_log_viewer_content_includes_prompt_and_graph_manifest(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1")
+        node_exec_id = self.conn.execute(
+            "SELECT node_exec_id FROM work_node_execution WHERE node_id = ?",
+            (work_graph_plan.PLANNING_NODE_ID,),
+        ).fetchone()[0]
+        log = execution_log.get_log(self.conn, node_exec_id)
+        self.assertIn("node_start", log["content"])
+        self.assertIn("node_completed", log["content"])
+        self.assertIn("实现贪吃蛇游戏", log["content"])
+        self.assertIn("n1", log["content"])  # a real node_id from _MANIFEST, proving the manifest is in there
+
+    def test_task_query_prepends_planning_node_as_the_first_node(self):
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1")
+
+        task = task_query.get_task(self.conn, "task-1")
+        display_nodes = task["work_graph_runs"][0]["graph_manifest"]["nodes"]
+        self.assertEqual(display_nodes[0]["node_id"], work_graph_plan.PLANNING_NODE_ID)
+        self.assertEqual([n["node_id"] for n in display_nodes[1:]], [n["node_id"] for n in _MANIFEST["nodes"]])
+        # the synthetic node is clickable exactly like any other -- it has a
+        # matching work_node_execution row in "nodes" with a real node_exec_id.
+        wne_by_id = {n["node_id"]: n for n in task["work_graph_runs"][0]["nodes"]}
+        self.assertIn(work_graph_plan.PLANNING_NODE_ID, wne_by_id)
+        self.assertTrue(wne_by_id[work_graph_plan.PLANNING_NODE_ID]["node_exec_id"])
+
+    def test_append_mode_does_not_record_a_second_planning_node(self):
+        # YAGNI (see _record_planning_node's own docstring): only the fresh
+        # branch gets one. Confirms append mode doesn't crash or duplicate.
+        with patch("src.work_graph_plan.subprocess.run") as run:
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
+            )
+            first = work_graph_plan.plan_work_graph(
+                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
+            )
+            run.return_value = _completed_process(
+                stdout_obj={"is_error": False, "result": _fenced_result(
+                    {"nodes": [{"node_id": "n5", "artifact_type": "test_dataset", "depends_on": []}], "edges": []}
+                )}
+            )
+            work_graph_plan.plan_work_graph(
+                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id=first["run_id"],
+            )
+        count = self.conn.execute(
+            "SELECT COUNT(*) FROM work_node_execution WHERE run_id = ? AND node_id = ?",
+            (first["run_id"], work_graph_plan.PLANNING_NODE_ID),
+        ).fetchone()[0]
+        self.assertEqual(count, 1)
+
+
 if __name__ == "__main__":
     unittest.main()