From cac67c9ab770dd904f019a6a5c14e2d697c8f247 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Sat, 15 Aug 2026 10:36:15 +0800 Subject: [PATCH] test: cover the planning-call work_node_execution recorded by _record_planning_node --- src/test_work_graph_plan.py | 105 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 104 insertions(+), 1 deletion(-) diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index aa6ae76..b5ae0be 100644 --- a/src/test_work_graph_plan.py +++ b/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() -- libgit2 0.21.4