宜博 / lemongraph

"""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"})

    def test_non_ascii_content_written_as_real_utf8_not_escaped(self):
        # User-reported (2026-08-13): v_log_viewer reads this .jsonl file's
        # raw bytes back and displays each line AS-IS (never JSON.parse()s
        # it), unlike every other json.dumps() call in this codebase whose
        # output always gets re-parsed before display. Without
        # ensure_ascii=False, real Chinese content here showed up in the
        # console as literal \uXXXX escape sequences instead of the actual
        # characters.
        manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []}
        with patch.object(
            org_node_trigger, "trigger_org_node_structured", return_value={"content": "期望行为：真实中文内容"}
        ):
            wgo.run_graph(
                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
            )

        raw_content = execution_log.get_log(self.conn, self._node_exec_id())["content"]
        self.assertIn("期望行为：真实中文内容", raw_content)
        self.assertNotIn("\\u671f", raw_content)  # 期 == 期, the escaped form this bug produced

        # 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()