"""U-031 tests: work_graph_orchestrate.py -- the graph_manifest -> per-node execution bridge closed during the REQ-01~14 regression audit. Real sqlite, run_fn mocked at the claude subprocess boundary (execute_work_graph is tested for real elsewhere, in test_work_graph_execute.py / the 2026-08-01 spike -- this module's own job is orchestration, not re-proving that). Run: python3 -m unittest src.test_work_graph_orchestrate -v """ import json import os import subprocess import tempfile import threading import time import unittest from unittest.mock import patch from src import ( execution_log, org_node_trigger, paths, schema, skill_query, skill_update, work_graph_orchestrate as wgo, work_node_run, ) class TopologicalOrderTest(unittest.TestCase): def test_linear_chain_via_depends_on(self): manifest = { "nodes": [ {"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}, {"node_id": "sdd", "depends_on": ["prd"]}, ], "edges": [], } self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["brd", "prd", "sdd"]) def test_linear_chain_via_edges_only(self): # real LLM output might populate edges but not each node's own # depends_on -- must still order correctly manifest = { "nodes": [{"node_id": "a"}, {"node_id": "b"}, {"node_id": "c"}], "edges": [{"from": "a", "to": "b"}, {"from": "b", "to": "c"}], } self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b", "c"]) def test_diamond_fan_out_fan_in(self): manifest = { "nodes": [ {"node_id": "n1", "depends_on": []}, {"node_id": "n2", "depends_on": ["n1"]}, {"node_id": "n3", "depends_on": ["n1"]}, {"node_id": "n4", "depends_on": ["n2", "n3"]}, ], "edges": [], } order = [n["node_id"] for n in wgo.topological_order(manifest)] self.assertEqual(order[0], "n1") self.assertEqual(order[-1], "n4") self.assertEqual(set(order[1:3]), {"n2", "n3"}) def test_cycle_raises(self): manifest = { "nodes": [{"node_id": "x", "depends_on": ["y"]}, {"node_id": "y", "depends_on": ["x"]}], "edges": [], } with self.assertRaises(ValueError): wgo.topological_order(manifest) def test_independent_nodes_ordered_deterministically(self): manifest = {"nodes": [{"node_id": "b"}, {"node_id": "a"}], "edges": []} self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b"]) class TopologicalLayersTest(unittest.TestCase): """G: topological_layers() -- the layer-grouped algorithm run_graph()'s concurrent scheduler is built on. topological_order() is now a thin flattening wrapper over this (tested above, unchanged behavior).""" def test_diamond_produces_three_layers(self): manifest = { "nodes": [ {"node_id": "n1", "depends_on": []}, {"node_id": "n2", "depends_on": ["n1"]}, {"node_id": "n3", "depends_on": ["n1"]}, {"node_id": "n4", "depends_on": ["n2", "n3"]}, ], "edges": [], } layers = wgo.topological_layers(manifest) self.assertEqual( [[n["node_id"] for n in layer] for layer in layers], [["n1"], ["n2", "n3"], ["n4"]], ) def test_linear_chain_is_all_single_node_layers(self): manifest = { "nodes": [ {"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}, {"node_id": "sdd", "depends_on": ["prd"]}, ], "edges": [], } layers = wgo.topological_layers(manifest) self.assertEqual( [[n["node_id"] for n in layer] for layer in layers], [["brd"], ["prd"], ["sdd"]], ) def test_independent_nodes_are_a_single_layer(self): manifest = {"nodes": [{"node_id": "b"}, {"node_id": "a"}], "edges": []} layers = wgo.topological_layers(manifest) self.assertEqual(len(layers), 1) self.assertEqual([n["node_id"] for n in layers[0]], ["a", "b"]) def test_cycle_raises(self): manifest = { "nodes": [{"node_id": "x", "depends_on": ["y"]}, {"node_id": "y", "depends_on": ["x"]}], "edges": [], } with self.assertRaises(ValueError): wgo.topological_layers(manifest) def test_topological_order_matches_flattened_layers(self): manifest = { "nodes": [ {"node_id": "n1", "depends_on": []}, {"node_id": "n2", "depends_on": ["n1"]}, {"node_id": "n3", "depends_on": ["n1"]}, ], "edges": [], } layers = wgo.topological_layers(manifest) flat = [n["node_id"] for layer in layers for n in layer] self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], flat) def test_on_failure_edge_does_not_create_a_spurious_cycle(self): # 2026-08-08: an on_failure edge commonly points BACKWARD (from a # node to one of its own upstream dependencies) -- the old # unfiltered edge merge would have folded this in as a (reversed, # incorrect) data dependency, creating a real cycle with the # existing forward depends_on and making this raise ValueError. manifest = { "nodes": [ {"node_id": "sdd", "depends_on": []}, {"node_id": "benchmark", "depends_on": ["sdd"]}, ], "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}], } layers = wgo.topological_layers(manifest) self.assertEqual([[n["node_id"] for n in layer] for layer in layers], [["sdd"], ["benchmark"]]) class TransitiveDependentsTest(unittest.TestCase): def test_includes_self_and_all_downstream(self): manifest = { "nodes": [ {"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": ["a"]}, {"node_id": "c", "depends_on": ["b"]}, {"node_id": "d", "depends_on": []}, # unrelated branch ], "edges": [], } self.assertEqual(wgo._transitive_dependents(manifest, "a"), {"a", "b", "c"}) self.assertEqual(wgo._transitive_dependents(manifest, "c"), {"c"}) self.assertEqual(wgo._transitive_dependents(manifest, "d"), {"d"}) def test_ignores_on_failure_edges(self): manifest = { "nodes": [ {"node_id": "sdd", "depends_on": []}, {"node_id": "benchmark", "depends_on": ["sdd"]}, ], "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}], } # sdd's dependents must be {sdd, benchmark} via the real depends_on # edge, not affected by the on_failure edge going the other way. self.assertEqual(wgo._transitive_dependents(manifest, "sdd"), {"sdd", "benchmark"}) class TransitiveAncestorsTest(unittest.TestCase): def test_includes_self_and_all_upstream(self): manifest = { "nodes": [ {"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": ["a"]}, {"node_id": "c", "depends_on": ["b"]}, ], "edges": [], } self.assertEqual(wgo._transitive_ancestors(manifest, "c"), {"a", "b", "c"}) self.assertEqual(wgo._transitive_ancestors(manifest, "a"), {"a"}) class OnFailureTargetTest(unittest.TestCase): def test_finds_the_on_failure_edge_target(self): manifest = { "nodes": [{"node_id": "sdd"}, {"node_id": "benchmark"}], "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}], } self.assertEqual(wgo._on_failure_target(manifest, "benchmark"), "sdd") def test_missing_edge_raises(self): manifest = {"nodes": [{"node_id": "benchmark"}], "edges": []} with self.assertRaises(ValueError): wgo._on_failure_target(manifest, "benchmark") class RunGraphTestBase(unittest.TestCase): def setUp(self): # DU-01: run_fn now really calls execution_log.append_log(), which # writes to paths.log_file_path() (default ~/.graphlemon/logs/...) -- # isolate that under a throwaway tempdir so real (non-mocked) # run_work_node/run_fn calls in this test module never touch the # developer's actual home directory, same pattern as test_paths.py. self.home_tmp = tempfile.TemporaryDirectory() self.addCleanup(self.home_tmp.cleanup) self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name self.addCleanup(self._restore_graphlemon_home) # G: check_same_thread=False -- run_graph()'s concurrent layers call # into this conn (via _LockedConn) from ThreadPoolExecutor worker # threads, not just the thread that created the connection. self.conn = schema.connect(":memory:", check_same_thread=False) schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')" ) self.conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', '{}', 'running')" ) self.conn.commit() def _restore_graphlemon_home(self): if self.old_graphlemon_home is None: os.environ.pop("GRAPHLEMON_HOME", None) else: os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home def tearDown(self): self.conn.close() _OUT_SCHEMA = {"type": "object", "required": ["content"]} _IN_SCHEMA = {"type": "object", "required": ["node_id"]} class RunGraphHappyPathTest(RunGraphTestBase): def test_runs_nodes_in_order_and_returns_outputs(self): manifest = { "nodes": [ {"node_id": "brd", "artifact_type": "brd", "depends_on": []}, {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, ], "edges": [], } calls = [] def build_prompt(node, outputs): calls.append((node["node_id"], dict(outputs))) return "prompt for " + node["node_id"] with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: mock_trigger.side_effect = [{"content": "brd text"}, {"content": "prd text"}] outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(outputs, {"brd": {"content": "brd text"}, "prd": {"content": "prd text"}}) self.assertEqual(calls[0], ("brd", {})) self.assertEqual(calls[1], ("prd", {"brd": {"content": "brd text"}})) def test_creates_work_node_execution_rows_and_placeholder_skill(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) row = self.conn.execute( "SELECT status, subagent_type, skill_ref FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'" ).fetchone() self.assertEqual(row, ("done", "planner", "planner-doc-writing")) skill_row = self.conn.execute("SELECT owner_org_node FROM skill WHERE skill_id = 'planner-doc-writing'").fetchone() self.assertEqual(skill_row[0], "planner") def test_resumes_without_rerunning_done_nodes(self): manifest = { "nodes": [ {"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}, ], "edges": [], } calls = [] def build_prompt(node, outputs): return "p-" + node["node_id"] with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: mock_trigger.side_effect = ( lambda org_node, prompt, cwd, schema, model=None, timeout=300: calls.append(prompt) or {"content": "c-" + prompt} ) wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(len(calls), 2) # simulate a restart: run_graph called again for the same run_id with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger2: mock_trigger2.side_effect = AssertionError("should not be called -- both nodes are already done") outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(set(outputs.keys()), {"brd", "prd"}) class RunGraphUsageLimitTest(RunGraphTestBase): def test_usage_limit_reached_translates_to_quota_wait(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} attempts = {"n": 0} def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): attempts["n"] += 1 if attempts["n"] == 1: raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "recovered"} sleeps = [] with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, sleep=lambda s: sleeps.append(s), now=lambda: 0.0, ) self.assertEqual(outputs["brd"], {"content": "recovered"}) self.assertEqual(sleeps, [5 * 60]) class RunGraphClassifierBlockedTest(RunGraphTestBase): def test_classifier_blocked_marks_node_blocked_and_propagates(self): # trigger_org_node_structured doesn't replicate execute_work_graph's # ClassifierBlockedError/permission_denials-threshold detection (no # call site ever branched on that specific type -- _make_run_fn's # own except-clause always treated it the same as any other generic # failure) -- any non-usage-limit OrgNodeTriggerError exercises the # exact same "mark node blocked and propagate" path. manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object( org_node_trigger, "trigger_org_node_structured", side_effect=org_node_trigger.OrgNodeTriggerError("blocked by classifier"), ): with self.assertRaises(org_node_trigger.OrgNodeTriggerError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) node_exec_id = self.conn.execute( "SELECT node_exec_id FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'" ).fetchone()[0] status = self.conn.execute( "SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,) ).fetchone()[0] self.assertEqual(status, "blocked") class RunGraphUnanticipatedExceptionMarksRunBlockedTest(RunGraphTestBase): """2026-08-08: found via a real self-planned run -- a bare exception from build_prompt() (a caller bug, not one of the three specifically- handled failure paths: loop_back limit exceeded, on_fail='hard_fail', legacy _verify_and_maybe_regenerate double-failure) used to propagate straight out of run_graph() leaving work_graph_run stuck at its non-terminal status forever, permanently blocking every future retry for this (task_id, org_node) via ux_work_graph_run_active's UNIQUE index -- the real incident this test reproduces.""" def test_build_prompt_keyerror_still_marks_run_blocked(self): manifest = {"nodes": [{"node_id": "sdd", "depends_on": []}], "edges": []} def bad_build_prompt(node, outputs): raise KeyError("prd_content") with patch.object(org_node_trigger, "trigger_org_node_structured"): with self.assertRaises(KeyError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, bad_build_prompt, _IN_SCHEMA, _OUT_SCHEMA, ) status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(status, "blocked") def test_already_terminal_status_is_left_alone(self): # if some inner handler already set a more specific terminal # status before the exception reaches run_graph()'s own catch-all, # the catch-all's WHERE clause must be a no-op, not overwrite it. self.conn.execute("UPDATE work_graph_run SET status = 'done' WHERE run_id = 'run-1'") self.conn.commit() manifest = {"nodes": [{"node_id": "sdd", "depends_on": []}], "edges": []} def bad_build_prompt(node, outputs): raise RuntimeError("boom") with patch.object(org_node_trigger, "trigger_org_node_structured"): with self.assertRaises(RuntimeError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, bad_build_prompt, _IN_SCHEMA, _OUT_SCHEMA, ) status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(status, "done") class VerifyFnTest(RunGraphTestBase): """REQ-02 boundary: verify_fn intercepts a bad node output before it propagates downstream, via one regeneration attempt.""" def test_verify_fn_passing_immediately_does_not_regenerate(self): manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []} with patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"content": "consistent prd"} ) as mock_trigger: outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=lambda node, output: (True, None), ) self.assertEqual(outputs["prd"], {"content": "consistent prd"}) self.assertEqual(mock_trigger.call_count, 1) run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(run_status, "running") # untouched -- verify_fn never failed def test_verify_fn_failing_once_regenerates_and_succeeds(self): manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []} contents = iter(["contradictory prd", "fixed prd"]) checked = [] def verify_fn(node, output): checked.append(output["content"]) return output["content"] == "fixed prd", ["scenario A contradicts scenario B"] with patch.object( org_node_trigger, "trigger_org_node_structured", side_effect=lambda *a, **k: {"content": next(contents)} ): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=verify_fn, ) self.assertEqual(outputs["prd"], {"content": "fixed prd"}) self.assertEqual(checked, ["contradictory prd", "fixed prd"]) run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(run_status, "verifying") reflect_row = self.conn.execute( "SELECT status FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'prd-reflect'" ).fetchone() self.assertEqual(reflect_row[0], "done") def test_verify_fn_failing_twice_raises_and_blocks_run(self): manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []} with patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"content": "still contradictory"} ): with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=lambda node, output: (False, ["still broken"]), ) run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(run_status, "blocked") class SelfTargetForcesWorktreeTest(RunGraphTestBase): """U-035 (05 §5.8): self-target unconditionally forces needs_worktree=True, regardless of what the caller passed. run_work_node itself is mocked here (not just trigger_org_node_structured) so this only asserts what needs_worktree value run_graph computes and passes down -- it doesn't exercise real git worktree commands against this actual checkout.""" def test_self_target_forces_worktree_even_when_caller_passed_false(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} self.conn.execute( "UPDATE project SET target_folder_path = ? WHERE project_id = 'proj-1'", (paths.repo_root(),) ) self.conn.commit() with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", paths.repo_root(), manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False, ) self.assertTrue(mock_run.call_args.kwargs["needs_worktree"]) def test_non_self_target_respects_caller_supplied_false(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False, ) self.assertFalse(mock_run.call_args.kwargs["needs_worktree"]) def test_non_self_target_still_respects_caller_supplied_true(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=True, ) self.assertTrue(mock_run.call_args.kwargs["needs_worktree"]) class RunFnLogsExecutionEventsTest(RunGraphTestBase): """DU-01 (bug A1/A2 root cause): `_make_run_fn()`'s `run_fn` closure must actually call `execution_log.append_log()` around the real `trigger_org_node_structured()` call, so `execution_log_file`'s backing file on disk genuinely has content -- not just a DB row tracking its lifecycle.""" def _node_exec_id(self, node_id="brd"): return self.conn.execute( "SELECT node_exec_id FROM work_node_execution WHERE run_id = 'run-1' AND node_id = ?", (node_id,) ).fetchone()[0] def _log_lines(self, node_exec_id): content = execution_log.get_log(self.conn, node_exec_id)["content"] return [json.loads(line) for line in content.splitlines() if line] def test_success_writes_node_start_then_node_completed(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "brd text"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) events = self._log_lines(self._node_exec_id()) self.assertEqual([e["event"] for e in events], ["node_start", "node_completed"]) self.assertEqual(events[1]["output"], {"content": "brd text"}) # also exercise the real HTTP-facing read path end to end (not just # the module function) -- an empty file was exactly bug A1. row = self.conn.execute( "SELECT size_bytes FROM execution_log_file WHERE node_exec_id = ?", (self._node_exec_id(),) ).fetchone() self.assertGreater(row[0], 0) def test_classifier_blocked_writes_node_failed_and_still_propagates(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} with patch.object( org_node_trigger, "trigger_org_node_structured", side_effect=org_node_trigger.OrgNodeTriggerError("blocked by classifier"), ): with self.assertRaises(org_node_trigger.OrgNodeTriggerError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) events = self._log_lines(self._node_exec_id()) self.assertEqual([e["event"] for e in events], ["node_start", "node_failed"]) self.assertIn("blocked by classifier", events[1]["error"]) def test_usage_limit_writes_quota_limit_then_retries(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} attempts = {"n": 0} def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): attempts["n"] += 1 if attempts["n"] == 1: raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "recovered"} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, sleep=lambda s: None, now=lambda: 0.0, ) self.assertEqual(outputs["brd"], {"content": "recovered"}) events = self._log_lines(self._node_exec_id()) self.assertEqual( [e["event"] for e in events], ["node_start", "quota_limit", "node_start", "node_completed"] ) self.assertEqual(events[1]["quota_limit_type"], "session") class RunGraphConcurrencyTest(RunGraphTestBase): """G, design doc §10 acceptance criteria 1-2: same-layer nodes actually run concurrently (overlapping wall-clock intervals), and the next layer never starts until every node in the previous layer has finished.""" def test_same_layer_nodes_run_concurrently(self): manifest = { "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}], "edges": [], } events = [] events_lock = threading.Lock() def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): start = time.monotonic() time.sleep(0.05) end = time.monotonic() with events_lock: events.append((start, end)) return {"content": prompt} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p-" + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(len(events), 2) (start_a, end_a), (start_b, end_b) = events self.assertLess(start_a, end_b) self.assertLess(start_b, end_a) def test_next_layer_waits_for_entire_previous_layer(self): manifest = { "nodes": [ {"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}, {"node_id": "c", "depends_on": ["a", "b"]}, ], "edges": [], } events = [] events_lock = threading.Lock() def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): start = time.monotonic() time.sleep(0.03) end = time.monotonic() with events_lock: events.append((start, end)) return {"content": "x"} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(len(events), 3) layer1_ends = [end for _, end in events[:2]] layer2_start, _ = events[2] self.assertGreaterEqual(layer2_start, max(layer1_ends)) class RunGraphLayerErrorTest(RunGraphTestBase): """G, design doc §10 acceptance criterion (FR-5): one node failing in a concurrent layer must not cancel an already-running sibling, and all failures in the layer are aggregated into one WorkGraphLayerError.""" def test_one_failing_node_does_not_cancel_sibling_and_aggregates_errors(self): manifest = { "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}], "edges": [], } calls = [] def fake_run_work_node(conn, node_exec_id, input_payload, input_schema, output_schema, run_fn, **kwargs): calls.append(input_payload["node_id"]) if input_payload["node_id"] == "a": raise RuntimeError("boom on a") return {"content": "b done"} with patch.object(work_node_run, "run_work_node", side_effect=fake_run_work_node): with self.assertRaises(wgo.WorkGraphLayerError) as ctx: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(sorted(calls), ["a", "b"]) # b was actually attempted, not cancelled error_node_ids = [node_id for node_id, exc in ctx.exception.errors] self.assertEqual(error_node_ids, ["a"]) class RunGraphConcurrentResumeTest(RunGraphTestBase): """G, design doc §10 acceptance criterion 3: breakpoint-recovery still works when the interrupted run had a concurrent (2+ node) layer -- a second run_graph() call for the same run_id must not re-run nodes the first call already finished.""" def test_concurrent_layer_resume_reruns_nothing(self): manifest = { "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}], "edges": [], } calls = [] def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): calls.append(prompt) return {"content": "c-" + prompt} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(len(calls), 2) with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger2: mock_trigger2.side_effect = AssertionError("should not be called -- both nodes already done") outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(set(outputs.keys()), {"a", "b"}) class RunGraphQuotaCoexistsWithNormalCompletionTest(RunGraphTestBase): """G, design doc §10 acceptance criterion 6: a node stuck on a quota-wait timeout in a concurrent layer must not lose its sibling's real, already- persisted output -- the sibling's row is 'done' with its real payload even though the overall layer raises.""" def test_quota_timeout_on_one_node_does_not_erase_siblings_persisted_output(self): manifest = { "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}], "edges": [], } def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): if prompt == "p-a": raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "b done"} now_calls = {"n": 0} now_lock = threading.Lock() def fake_now(): # first call records quota_wait_started_at as "0"; second call # (computing elapsed) reports far past the session timeout, so # this times out immediately instead of really sleeping. with now_lock: now_calls["n"] += 1 return 0.0 if now_calls["n"] == 1 else 999999.0 with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): with self.assertRaises(wgo.WorkGraphLayerError) as ctx: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p-" + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, sleep=lambda s: None, now=fake_now, ) error_node_ids = [node_id for node_id, exc in ctx.exception.errors] self.assertEqual(error_node_ids, ["a"]) self.assertIsInstance(dict(ctx.exception.errors)["a"], work_node_run.UsageLimitWaitTimeoutError) b_row = self.conn.execute( "SELECT status, output_payload FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'b'" ).fetchone() self.assertEqual(b_row[0], "done") self.assertEqual(json.loads(b_row[1]), {"content": "b done"}) class RunGraphRealSqliteConcurrencyTest(unittest.TestCase): """G, design doc §10 acceptance criterion 4: a real (non-:memory:, non-mocked-conn) sqlite database under genuine concurrent write pressure from a multi-node layer must never raise 'database is locked' -- run several rounds since races like this often only reproduce probabilistically.""" def setUp(self): self.home_tmp = tempfile.TemporaryDirectory() self.addCleanup(self.home_tmp.cleanup) self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name self.addCleanup(self._restore_graphlemon_home) self.db_tmp = tempfile.TemporaryDirectory() self.addCleanup(self.db_tmp.cleanup) def _restore_graphlemon_home(self): if self.old_graphlemon_home is None: os.environ.pop("GRAPHLEMON_HOME", None) else: os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home def test_multiple_rounds_of_concurrent_layer_writes_never_raise_locked(self): manifest = { "nodes": [{"node_id": n, "depends_on": []} for n in ("a", "b", "c", "d")], "edges": [], } def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): time.sleep(0.005) return {"content": prompt} for round_n in range(5): db_path = os.path.join(self.db_tmp.name, "test_{}.db".format(round_n)) conn = schema.connect(db_path, check_same_thread=False) schema.create_schema(conn) conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')" ) conn.execute( "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) " "VALUES ('run-1', 'task-1', '{}', 'running')" ) conn.commit() try: with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(set(outputs.keys()), {"a", "b", "c", "d"}) finally: conn.close() class CoerceBoolTest(unittest.TestCase): def test_real_bool_passes_through(self): self.assertTrue(wgo._coerce_bool(True)) self.assertFalse(wgo._coerce_bool(False)) def test_string_false_is_not_truthy(self): # regression: a real call has returned "false" as a JSON STRING # despite "type": "boolean" in the schema -- `if "false":` is # truthy in Python (same class of bug e2e_pipeline._coerce_bool # was written for). self.assertFalse(wgo._coerce_bool("false")) def test_string_true_is_treated_as_true(self): self.assertTrue(wgo._coerce_bool("true")) class ResolveNodeVerifyFnTest(unittest.TestCase): """2026-08-08: node-declared `judgment` takes over verify_fn resolution; a node with no `judgment` falls through to whatever the caller passed as `verify_fn` (FR-A8 -- every pre-existing graph_manifest has no `judgment` field at all).""" def test_no_judgment_falls_through_to_external_verify_fn(self): external = lambda node, output: (True, None) resolved = wgo._resolve_node_verify_fn({"node_id": "x"}, external, "planner", "/repo", None) self.assertIs(resolved, external) def test_judgment_kind_none_falls_through_to_external_verify_fn(self): external = lambda node, output: (True, None) node = {"node_id": "x", "judgment": {"kind": "none"}} resolved = wgo._resolve_node_verify_fn(node, external, "planner", "/repo", None) self.assertIs(resolved, external) def test_no_judgment_and_no_external_verify_fn_returns_none(self): resolved = wgo._resolve_node_verify_fn({"node_id": "x"}, None, "planner", "/repo", None) self.assertIsNone(resolved) def test_mechanical_judgment_dispatches_to_registered_check(self): node = {"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "my_check"}} registry = {"my_check": lambda content: (content == "good", None if content == "good" else "bad content")} resolved = wgo._resolve_node_verify_fn(node, None, "planner", "/repo", registry) ok, feedback = resolved(node, {"content": "good"}) self.assertTrue(ok) ok, feedback = resolved(node, {"content": "bad"}) self.assertFalse(ok) self.assertEqual(feedback, ["bad content"]) def test_mechanical_judgment_unknown_check_ref_raises(self): node = {"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "typo"}} with self.assertRaises(ValueError): wgo._resolve_node_verify_fn(node, None, "planner", "/repo", {"real_check": lambda c: (True, None)}) def test_unknown_judgment_kind_raises(self): node = {"node_id": "x", "judgment": {"kind": "astrology"}} with self.assertRaises(ValueError): wgo._resolve_node_verify_fn(node, None, "planner", "/repo", None) class ValidateMechanicalCheckRefsTest(unittest.TestCase): """Fail before running any node, not when the offending node is reached.""" def test_unregistered_check_ref_raises(self): manifest = { "nodes": [{"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "typo"}}], "edges": [], } with self.assertRaises(ValueError): wgo._validate_mechanical_check_refs(manifest, {"real_check": lambda c: (True, None)}) def test_registered_check_ref_passes(self): manifest = { "nodes": [{"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "real_check"}}], "edges": [], } wgo._validate_mechanical_check_refs(manifest, {"real_check": lambda c: (True, None)}) # must not raise def test_no_judgment_nodes_pass_with_no_registry(self): manifest = {"nodes": [{"node_id": "brd"}], "edges": []} wgo._validate_mechanical_check_refs(manifest, None) # must not raise class LlmJudgmentVerifyFnTest(unittest.TestCase): def test_non_self_target_calls_directly_without_worktree(self): judgment = {"judge_prompt": "Review: {content}"} verify_fn = wgo._llm_judgment_verify_fn(judgment, "planner", "/repo") with patch.object(wgo.paths, "is_self_target", return_value=False), \ patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"ok": True, "issues": []}, ) as mock_trigger: ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) self.assertTrue(ok) self.assertEqual(issues, []) self.assertEqual(mock_trigger.call_args[0][0], "planner") self.assertEqual(mock_trigger.call_args[0][2], "/repo") def test_string_ok_is_coerced(self): judgment = {"judge_prompt": "Review: {content}"} verify_fn = wgo._llm_judgment_verify_fn(judgment, "planner", "/repo") with patch.object(wgo.paths, "is_self_target", return_value=False), \ patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"ok": "false", "issues": ["bad"]}, ): ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) self.assertFalse(ok) self.assertEqual(issues, ["bad"]) def test_self_target_isolates_in_its_own_worktree(self): # same pattern as e2e_pipeline.py's PrdConsistencySelfTargetWorktreeTest judgment = {"judge_prompt": "Review: {content}"} verify_fn = wgo._llm_judgment_verify_fn(judgment, "planner", "/repo") with patch.object(wgo.paths, "is_self_target", return_value=True), \ patch.object(wgo.work_node_run, "create_worktree", return_value="/repo/.graphlemon-worktrees/x") as mock_create, \ patch.object(wgo.work_node_run, "ensure_worktree_committed") as mock_ensure, \ patch.object(wgo.work_node_run, "merge_worktree") as mock_merge, \ patch.object(wgo.work_node_run, "remove_worktree") as mock_remove, \ patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"ok": True, "issues": []}, ) as mock_trigger: ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) self.assertTrue(ok) mock_create.assert_called_once_with("/repo", unittest.mock.ANY) self.assertEqual(mock_trigger.call_args[0][2], "/repo/.graphlemon-worktrees/x") self.assertEqual(mock_ensure.call_count, 2) mock_merge.assert_called_once() mock_remove.assert_called_once() class RunGraphMechanicalJudgmentTest(RunGraphTestBase): """End-to-end: a node's own judgment field (not an externally-passed verify_fn) drives verification through run_graph().""" def test_node_declared_mechanical_judgment_regenerates_on_failure(self): manifest = { "nodes": [{"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "compiles"}}], "edges": [], } registry = {"compiles": lambda content: (content == "real code", None if content == "real code" else "not code")} with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: mock_trigger.side_effect = [{"content": "narrative text"}, {"content": "real code"}] outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["script"], {"content": "real code"}) def test_unregistered_check_ref_fails_before_any_node_runs(self): manifest = { "nodes": [{"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "typo"}}], "edges": [], } with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: with self.assertRaises(ValueError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry={"compiles": lambda c: (True, None)}, ) mock_trigger.assert_not_called() class RunGraphLoopBackTest(RunGraphTestBase): """2026-08-08, Work Graph self-planning + loop-back design §4.""" def test_retry_self_regenerates_with_accumulating_history(self): # A node with on_fail.action='retry_self' loops back to itself -- # fails twice, succeeds the third time. Each retry's prompt must # include the FULL history of prior rejections, not just the last one. manifest = { "nodes": [{ "node_id": "benchmark", "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "retry_self", "max_attempts": 5}, }], "edges": [], } contents = ["bad v1", "bad v2", "good"] prompts_seen = [] def build_prompt(node, outputs): return "base prompt" def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): prompts_seen.append(prompt) return {"content": contents.pop(0)} registry = {"compiles": lambda content: (content == "good", "not real code: " + content)} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["benchmark"], {"content": "good"}) # 3 real executions happened (2 failures + 1 success) -- 3 distinct # work_node_execution rows, not 1 (which would mean a stale 'done' # row got silently reused instead of genuinely re-running). rows = self.conn.execute( "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' ORDER BY node_id" ).fetchall() self.assertEqual([r[0] for r in rows], ["benchmark", "benchmark-attempt2", "benchmark-attempt3"]) events = self.conn.execute( "SELECT payload FROM execution_event WHERE event_type = 'loop_back_triggered' ORDER BY created_at" ).fetchall() self.assertEqual(len(events), 2) payloads = [json.loads(p) for (p,) in events] self.assertEqual(payloads[0]["attempt_number"], 1) self.assertEqual(payloads[1]["attempt_number"], 2) # the THIRD (successful) execution's prompt must carry both prior # rejections' feedback, not just the most recent one -- inspect what # got passed as trigger_org_node_structured's own `prompt` argument # directly (input_payload is just {"node_id": ...}, the real prompt # text never round-trips through the DB). third_call_prompt = prompts_seen[2] self.assertIn("not real code: bad v1", third_call_prompt) self.assertIn("not real code: bad v2", third_call_prompt) def test_loop_back_to_earlier_node_reruns_target_and_downstream(self): # benchmark declares on_fail.action='loop_back' with an on_failure # edge back to sdd -- sdd must be genuinely re-executed (not # short-circuited to its stale 'done' output), and benchmark's # second attempt must see the NEW sdd output. manifest = { "nodes": [ {"node_id": "sdd", "depends_on": []}, { "node_id": "benchmark", "depends_on": ["sdd"], "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "loop_back", "max_attempts": 3}, }, ], "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}], } sdd_outputs = ["sdd v1", "sdd v2 (fixed)"] benchmark_seen_sdd = [] def build_prompt(node, outputs): if node["node_id"] == "benchmark": benchmark_seen_sdd.append(outputs["sdd"]["content"]) return "benchmark prompt using " + outputs["sdd"]["content"] return "sdd prompt" # sdd runs before benchmark in each pass (topological order), so # call order alone deterministically identifies which node this is: # sdd v1, bad benchmark, sdd v2 (re-run via loop_back), good benchmark. call_order = [] def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): call_order.append(prompt) call_index = len(call_order) if call_index == 1: return {"content": sdd_outputs[0]} if call_index == 2: return {"content": "bad benchmark"} if call_index == 3: return {"content": sdd_outputs[1]} return {"content": "good benchmark"} registry = {"compiles": lambda content: (content == "good benchmark", "rejected: " + content)} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["sdd"], {"content": sdd_outputs[1]}) self.assertEqual(outputs["benchmark"], {"content": "good benchmark"}) self.assertEqual(benchmark_seen_sdd, [sdd_outputs[0], sdd_outputs[1]]) sdd_rows = self.conn.execute( "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' AND " "(node_id = 'sdd' OR node_id LIKE 'sdd-attempt%') ORDER BY node_id" ).fetchall() self.assertEqual([r[0] for r in sdd_rows], ["sdd", "sdd-attempt2"]) def test_exceeding_limit_raises_and_blocks_run(self): manifest = { "nodes": [{ "node_id": "benchmark", "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "retry_self", "max_attempts": 2}, }], "edges": [], } registry = {"compiles": lambda content: (False, "always rejected")} with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "always bad"}): with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(status, "blocked") def test_hard_fail_raises_immediately_with_no_retry(self): manifest = { "nodes": [{ "node_id": "benchmark", "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "hard_fail"}, }], "edges": [], } registry = {"compiles": lambda content: (False, "bad")} with patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"} ) as mock_trigger: with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(mock_trigger.call_count, 1) # no retry at all def test_resume_after_restart_continues_from_the_real_attempt_count(self): # Simulates a PRIOR process that ran attempt 1 (rejected), recorded # the loop_back event, then died before attempt 2 ever began -- # seeded directly via SQL rather than literally raising mid-flight # inside execute_work_graph, which would instead exercise a # separate, pre-existing gap in work_node_run.py's own resume # logic (it unconditionally INSERTs a new execution_log_file row # every call, with no existing-row check, so resuming a node_exec_id # that already has one hits a real UNIQUE constraint violation -- # a genuine bug, but in a different module and out of scope here). manifest = { "nodes": [{ "node_id": "benchmark", "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "retry_self", "max_attempts": 5}, }], "edges": [], } registry = {"compiles": lambda content: (content == "good", "bad: " + content)} wgo.ensure_placeholder_skill(self.conn, "planner-doc-writing", "planner") self.conn.execute( "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " "input_schema_ref, output_schema_ref, input_payload, output_payload, status) " "VALUES ('exec-1', 'run-1', 'benchmark', 'planner', 'planner-doc-writing', " "'work_graph_node_input.schema.json', 'work_graph_node_output.schema.json', " "'{}', '{\"content\": \"bad v1\"}', 'done')" ) self.conn.execute( "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) " "VALUES ('evt-1', 'proj-1', 'task-1', 'run-1', 'loop_back_triggered', ?)", (json.dumps({ "source_node_id": "benchmark", "target_node_id": "benchmark", "attempt_number": 1, "feedback": ["bad: bad v1"], }),), ) self.conn.commit() # "restart": a fresh run_graph() call for the same run_id, this # time succeeding -- must resume as attempt 2 (reading the true # count from execution_event), not restart from attempt 1 or lose # the earlier rejection's history. with patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"content": "good"} ) as mock_trigger: outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["benchmark"], {"content": "good"}) rows = self.conn.execute( "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' ORDER BY node_id" ).fetchall() self.assertEqual([r[0] for r in rows], ["benchmark", "benchmark-attempt2"]) self.assertIn("bad: bad v1", mock_trigger.call_args[0][1]) def test_org_node_type_scoped_config_limit_used_when_no_node_level_override(self): from src import settings_config settings_config.upsert_loop_limit_config(self.conn, "planner:benchmark", 1, "test") manifest = { "nodes": [{ "node_id": "benchmark", "artifact_type": "benchmark", "judgment": {"kind": "mechanical", "check_ref": "compiles"}, "on_fail": {"action": "retry_self"}, # no max_attempts -- falls back to config }], "edges": [], } registry = {"compiles": lambda content: (False, "bad")} with patch.object( org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"} ) as mock_trigger: with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) # max_attempts=1 means "1 retry allowed" (matches the legacy # exactly-one-retry default, design doc §2's on_fail comment) -- # attempt_number=1 on the first failure is not > 1, so one retry # happens; attempt_number=2 on THAT failure is > 1 and exceeds. # Two real executions total, not one. self.assertEqual(mock_trigger.call_count, 2) class NodeLevelSubagentSkillTest(RunGraphTestBase): """Phase 2 §4: a node can declare its own subagent_type/skill_ref, overriding the run-level org_node/skill_id defaults -- SDD §10 Phase 2a acceptance criterion 1.""" def test_node_level_fields_override_run_level_defaults(self): manifest = { "nodes": [ {"node_id": "a", "depends_on": [], "subagent_type": "coder", "skill_ref": "code-task-development"}, {"node_id": "b", "depends_on": []}, # no override -- uses run-level org_node/skill_id ], "edges": [], } with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) rows = { r[0]: (r[1], r[2]) for r in self.conn.execute( "SELECT node_id, subagent_type, skill_ref FROM work_node_execution WHERE run_id = 'run-1'" ).fetchall() } self.assertEqual(rows["a"], ("coder", "code-task-development")) self.assertEqual(rows["b"], ("planner", "planner-doc-writing")) def test_node_level_org_node_used_for_llm_judgment_agent_identity(self): # A node's own judgment.kind='llm' call must run under ITS OWN # subagent_type, not the run-level default -- otherwise a node # declaring subagent_type='validator' for judgment purposes would # silently get judged under the wrong persona. manifest = { "nodes": [{ "node_id": "a", "depends_on": [], "subagent_type": "coder", "judgment": {"kind": "llm", "judge_prompt": "check {content}"}, "on_fail": {"action": "hard_fail"}, }], "edges": [], } seen_org_nodes = [] def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): if "content" in output_schema.get("required", []): return {"content": "x"} seen_org_nodes.append(org_node) return {"ok": True, "issues": []} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(seen_org_nodes, ["coder"]) class DynamicLoopBackContentErrorTest(RunGraphTestBase): """Phase 2 §5: on_fail.action='dynamic_loop_back' -- Reflection-review picks a target node from the failing node's real ancestors and judges defect_kind. content_error skips Fix-Skill and loops back directly -- SDD §10 Phase 2a acceptance criterion 2.""" def test_unanimous_content_error_loops_back_to_chosen_ancestor(self): manifest = { "nodes": [ {"node_id": "sdd", "depends_on": []}, { "node_id": "final_review", "depends_on": ["sdd"], "judgment": {"kind": "mechanical", "check_ref": "is_good"}, "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, }, ], "edges": [], } sdd_contents = ["sdd v1", "sdd v2 (fixed)"] node_call_count = [0] lock = threading.Lock() def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): required = output_schema.get("required", []) if "target_node_id" in required: return {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "sdd is stale"} if "content" in required: with lock: node_call_count[0] += 1 idx = node_call_count[0] if idx == 1: return {"content": sdd_contents[0]} if idx == 2: return {"content": "bad final review"} if idx == 3: return {"content": sdd_contents[1]} return {"content": "good final review"} raise AssertionError("unexpected output_schema: " + repr(output_schema)) registry = {"is_good": lambda content: (content == "good final review", None if content == "good final review" else "rejected: " + content)} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "prompt for " + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["sdd"], {"content": sdd_contents[1]}) self.assertEqual(outputs["final_review"], {"content": "good final review"}) rows = [ r[0] for r in self.conn.execute( "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' ORDER BY node_id" ).fetchall() ] self.assertEqual(rows, ["final_review", "final_review-attempt2", "sdd", "sdd-attempt2"]) events = self.conn.execute( "SELECT payload FROM execution_event WHERE event_type = 'loop_back_triggered'" ).fetchall() self.assertEqual(len(events), 1) payload = json.loads(events[0][0]) self.assertEqual(payload["source_node_id"], "final_review") self.assertEqual(payload["target_node_id"], "sdd") # no skill_iterated event -- content_error must not touch any Skill. skill_events = self.conn.execute( "SELECT 1 FROM execution_event WHERE event_type = 'skill_iterated'" ).fetchall() self.assertEqual(skill_events, []) class TallyReflectionTest(unittest.TestCase): """_tally_reflection in isolation -- the majority-vote + degradation logic _judge_root_cause_node relies on. An invalid winner (not among candidate_node_ids) is corrected HERE, before _run_one_node's own 'target not in ancestors' runtime check (§5.7 point 1) would ever see it -- that check is defense-in-depth for if this guarantee is ever weakened, not the primary enforcement, so it's tested at this level.""" def test_unanimous_valid_winner(self): judgments = [ {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "a"}, {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "b"}, {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "c"}, ] result = wgo._tally_reflection(judgments, ["sdd", "prd"]) self.assertEqual(result["target_node_id"], "sdd") self.assertEqual(result["defect_kind"], "content_error") def test_two_thirds_skill_defect_majority(self): judgments = [ {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "a"}, {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "b"}, {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "c"}, ] result = wgo._tally_reflection(judgments, ["sdd", "prd"]) self.assertEqual(result["target_node_id"], "sdd") self.assertEqual(result["defect_kind"], "skill_defect") def test_winner_not_in_candidates_degrades_to_first_candidate_content_error(self): judgments = [ {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "a"}, {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "b"}, {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "c"}, ] result = wgo._tally_reflection(judgments, ["sdd", "prd"]) self.assertEqual(result["target_node_id"], "sdd") self.assertEqual(result["defect_kind"], "content_error") def test_no_two_thirds_agreement_degrades(self): judgments = [ {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "a"}, {"target_node_id": "prd", "defect_kind": "content_error", "detail": "b"}, {"target_node_id": "benchmark", "defect_kind": "content_error", "detail": "c"}, ] result = wgo._tally_reflection(judgments, ["sdd", "prd", "benchmark"]) self.assertEqual(result["defect_kind"], "content_error") self.assertIn(result["target_node_id"], ["sdd", "prd", "benchmark"]) def _init_git_repo(path): subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True) class DynamicLoopBackSkillDefectTest(RunGraphTestBase): """Phase 2 §5.5: defect_kind='skill_defect' triggers Fix-Skill -- a real skill_update.update_skill() call against a real (throwaway) git repo, never graphlemon's own checkout -- before the loop_back fires. SDD §10 Phase 2a acceptance criterion 3.""" def setUp(self): super().setUp() self.skill_repo = tempfile.TemporaryDirectory() self.addCleanup(self.skill_repo.cleanup) _init_git_repo(self.skill_repo.name) skill_update._write_and_commit(self.skill_repo.name, "skills/my-skill/SKILL.md", "v1 content", "seed skill") self.conn.execute( "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " "VALUES ('my-skill', 'coder', 'skills/my-skill', 'unknown', 'migrated_from_hermes', 'active')" ) self.conn.commit() self.repo_root_patch = patch.object(skill_query, "_REPO_ROOT", self.skill_repo.name) self.repo_root_patch.start() self.addCleanup(self.repo_root_patch.stop) def test_skill_defect_calls_real_update_skill_before_loop_back(self): manifest = { "nodes": [ {"node_id": "sdd", "depends_on": [], "skill_ref": "my-skill"}, { "node_id": "final_review", "depends_on": ["sdd"], "judgment": {"kind": "mechanical", "check_ref": "is_good"}, "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3}, }, ], "edges": [], } node_call_count = [0] lock = threading.Lock() def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): required = output_schema.get("required", []) if "target_node_id" in required: return {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "systemic issue"} if "new_content" in required: return {"new_content": "v2 content, fixed", "change_summary": "fixed the systemic issue"} if "content" in required: with lock: node_call_count[0] += 1 idx = node_call_count[0] if idx in (1, 3): return {"content": "sdd content"} if idx == 2: return {"content": "bad final review"} return {"content": "good final review"} raise AssertionError("unexpected output_schema: " + repr(output_schema)) registry = {"is_good": lambda content: (content == "good final review", None if content == "good final review" else "rejected: " + content)} with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "prompt for " + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) self.assertEqual(outputs["final_review"], {"content": "good final review"}) # a real skill_iteration row was produced, with a real (different) # after_git_commit_sha -- not a stub, an actual git commit. iteration = self.conn.execute( "SELECT skill_id, before_git_commit_sha, after_git_commit_sha, change_summary " "FROM skill_iteration" ).fetchone() self.assertEqual(iteration[0], "my-skill") self.assertNotEqual(iteration[1], iteration[2]) self.assertEqual(iteration[3], "fixed the systemic issue") with open(os.path.join(self.skill_repo.name, "skills/my-skill/SKILL.md")) as f: self.assertEqual(f.read(), "v2 content, fixed") # a real bug_record(root_cause_category='skill_defect') was created # and linked back to the iteration (skill_update.update_skill()'s # own contract). bug_row = self.conn.execute( "SELECT root_cause_category, skill_iteration_suggestion_id FROM bug_record" ).fetchone() self.assertEqual(bug_row[0], "skill_defect") self.assertIsNotNone(bug_row[1]) # skill_iterated execution_event recorded (existing but previously # never-written CHECK enum value). skill_events = self.conn.execute( "SELECT payload FROM execution_event WHERE event_type = 'skill_iterated'" ).fetchall() self.assertEqual(len(skill_events), 1) payload = json.loads(skill_events[0][0]) self.assertEqual(payload["node_id"], "sdd") self.assertEqual(payload["skill_id"], "my-skill") # graphlemon's own real checkout was never touched -- the Skill # write landed only in the throwaway skill_repo tempdir. self.assertFalse( os.path.exists(os.path.join(paths.repo_root(), "skills", "my-skill", "SKILL.md.orig")) ) def test_rate_limit_stops_repeated_rewrites_of_same_skill(self): # §5.7 point 5: after _MAX_SKILL_FIXES_PER_RUN_PER_SKILL rewrites of # the same (run_id, skill_id) this run, Fix-Skill silently skips -- # dynamic_loop_back still proceeds as a plain loop_back. for i in range(wgo._MAX_SKILL_FIXES_PER_RUN_PER_SKILL): wgo._log_skill_iterated_event(self.conn, "run-1", "sdd", "my-skill", "iteration-{}".format(i)) manifest = {"nodes": [{"node_id": "sdd", "depends_on": [], "skill_ref": "my-skill"}], "edges": []} result = wgo._fix_skill_for_node( self.conn, "run-1", "sdd", manifest, lambda node, outputs: "p", {}, ["feedback"], "/tmp/proj-1", ) self.assertIsNone(result) # no new skill_iteration row was created beyond what we seeded above. count = self.conn.execute("SELECT COUNT(*) FROM skill_iteration").fetchone()[0] self.assertEqual(count, 0) # the two seeded events above didn't insert real skill_iteration rows if __name__ == "__main__": unittest.main()