宜博 / lemongraph

"""U-011 tests: c_work_graph_plan, mocked at the subprocess boundary (CLAUDE.md
Testing constraint). Real-call coverage is in src/test_work_graph_plan_e2e.py
(RUN_REAL_AGENT_E2E=1) -- that real call is what proved --json-schema doesn't
constrain output when --agent is set, which is why these mocks return
free-form ```json-fenced text in `result`, not `structured_output`.

Run: python3 -m unittest src.test_work_graph_plan -v
"""

import json
import subprocess
import unittest
from unittest.mock import patch

from src import execution_log, org_node_trigger, schema, task_query, work_graph_orchestrate, work_graph_plan

_MANIFEST = {
    "nodes": [
        {"node_id": "n1", "artifact_type": "brd", "depends_on": []},
        {"node_id": "n2", "artifact_type": "prd", "depends_on": ["n1"]},
        {"node_id": "n3", "artifact_type": "sdd", "depends_on": ["n2"]},
        {"node_id": "n4", "artifact_type": "benchmark", "depends_on": ["n3"]},
    ],
    "edges": [
        {"from": "n1", "to": "n2", "edge_type": "needs_judgment"},
        {"from": "n2", "to": "n3", "edge_type": "needs_judgment"},
        {"from": "n3", "to": "n4", "edge_type": "needs_judgment"},
    ],
}


def _completed_process(returncode=0, stdout_obj=None, stderr=""):
    stdout = json.dumps(stdout_obj) if stdout_obj is not None else ""
    return subprocess.CompletedProcess(args=["claude"], returncode=returncode, stdout=stdout, stderr=stderr)


def _fenced_result(manifest, wrapper_key=None):
    body = {wrapper_key: manifest} if wrapper_key else manifest
    return "Here is my plan:\n\n```json\n" + json.dumps(body) + "\n```\n\nDone."


class PlanWorkGraphTestBase(unittest.TestCase):
    def setUp(self):
        # check_same_thread=False: GenericPathEndToEndViaRunGraphTest below
        # drives a real multi-node concurrent layer through run_graph()
        # (G), which calls into this conn from ThreadPoolExecutor worker
        # threads via _LockedConn -- see test_work_graph_orchestrate.py's
        # RunGraphTestBase for the same reasoning.
        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', 'queued', 'planner')"
        )
        self.conn.commit()

    def tearDown(self):
        self.conn.close()


class ExtractJsonPayloadTest(unittest.TestCase):
    def test_bare_json_result(self):
        payload = {"content": "hello"}
        self.assertEqual(work_graph_plan.extract_json_payload(json.dumps(payload), ["content"]), payload)

    def test_fenced_json_block_with_prose_around_it(self):
        payload = {"content": "hello"}
        text = "Here you go:\n\n```json\n" + json.dumps(payload) + "\n```\n\nDone."
        self.assertEqual(work_graph_plan.extract_json_payload(text, ["content"]), payload)

    def test_missing_required_key_returns_none(self):
        text = "```json\n" + json.dumps({"other": "x"}) + "\n```"
        self.assertIsNone(work_graph_plan.extract_json_payload(text, ["content"]))

    def test_no_json_anywhere_returns_none(self):
        self.assertIsNone(work_graph_plan.extract_json_payload("sorry, I can't do that", ["content"]))

    def test_content_containing_braces_extracts_correctly(self):
        # a real code-writing node's content is guaranteed to contain '{'/'}'
        # (dict literals, f-strings, function calls) -- must not confuse
        # extraction just because the payload itself contains braces.
        payload = {"content": 'def f():\n    d = {"a": 1, "b": 2}\n    return d\n'}
        text = "```json\n" + json.dumps(payload) + "\n```"
        self.assertEqual(work_graph_plan.extract_json_payload(text, ["content"]), payload)

    def test_content_with_brace_then_whitespace_then_backticks_does_not_truncate(self):
        # Real incident (2026-08-15, task b243dfe8), diagnosed live with the
        # user after they pushed back on "is this over-constrained code" --
        # the honest answer was the opposite: the OLD implementation
        # approximated JSON-boundary-finding with a non-greedy regex
        # (```(?:json)?\s*(\{.*?\})\s*```), which stops at the FIRST '}'
        # anywhere in the text followed by whitespace + a closing fence.
        # This is fully VALID, json.dumps()-produced JSON -- no malformed
        # input required -- the content just happens to contain a dict
        # literal followed by spaces (legal, unescaped JSON string content)
        # and then a literal ``` sequence, e.g. a script whose own docstring
        # shows an example. The regex silently returned None here; a real
        # JSON decoder (json.JSONDecoder.raw_decode, which this function now
        # uses) correctly walks past it because it understands JSON string/
        # brace grammar instead of approximating it.
        payload = {"content": 'd = {"a": 1}   ```\nmore real code follows this line\n'}
        text = "```json\n" + json.dumps(payload) + "\n```"
        self.assertEqual(work_graph_plan.extract_json_payload(text, ["content"]), payload)


class ExtractGraphManifestTest(unittest.TestCase):
    def test_bare_json_result(self):
        self.assertEqual(work_graph_plan._extract_graph_manifest(json.dumps(_MANIFEST)), _MANIFEST)

    def test_fenced_json_block_with_prose_around_it(self):
        self.assertEqual(work_graph_plan._extract_graph_manifest(_fenced_result(_MANIFEST)), _MANIFEST)

    def test_fenced_json_wrapped_under_graph_manifest_key(self):
        text = _fenced_result(_MANIFEST, wrapper_key="graph_manifest")
        self.assertEqual(work_graph_plan._extract_graph_manifest(text), _MANIFEST)

    def test_no_json_anywhere_returns_none(self):
        self.assertIsNone(work_graph_plan._extract_graph_manifest("sorry, I can't do that"))

    def test_wrapper_key_content_containing_braces_does_not_truncate(self):
        # Same class of bug as ExtractJsonPayloadTest's own brace/whitespace/
        # backtick case, for this function's OWN separate wrapper-key
        # fallback scan (previously also regex-based, same vulnerability).
        manifest_with_brace_content = {
            "nodes": [{
                "node_id": "n1", "artifact_type": "brd", "depends_on": [],
                "note": 'example: {"x": 1}   ```\nmore text\n',
            }],
            "edges": [],
        }
        text = _fenced_result(manifest_with_brace_content, wrapper_key="graph_manifest")
        self.assertEqual(work_graph_plan._extract_graph_manifest(text), manifest_with_brace_content)


class NormalizeGraphManifestTest(unittest.TestCase):
    """U-031 regression: a real Planner call (2026-08-01) came back using
    `id` instead of the documented `node_id`, and never populated any
    node's own `depends_on` -- dependencies existed only as `edges`. This
    is a real, observed deviation, not a hypothetical one; the fixture
    below is the actual shape captured from that call (fields trimmed to
    what normalization touches)."""

    def test_already_well_formed_manifest_untouched(self):
        manifest = {
            "nodes": [
                {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
                {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
            ],
            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        self.assertEqual(result["nodes"][1]["depends_on"], ["brd"])

    def test_id_key_normalized_to_node_id(self):
        manifest = {
            "nodes": [{"id": "brd", "artifact_type": "brd"}, {"id": "prd", "artifact_type": "prd"}],
            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        self.assertEqual([n["node_id"] for n in result["nodes"]], ["brd", "prd"])

    def test_depends_on_derived_from_edges_when_absent(self):
        manifest = {
            "nodes": [{"id": "brd"}, {"id": "prd"}, {"id": "sdd"}, {"id": "benchmark"}],
            "edges": [
                {"from": "brd", "to": "prd", "edge_type": "needs_judgment"},
                {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"},
                {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"},
            ],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        by_id = {n["node_id"]: n for n in result["nodes"]}
        self.assertEqual(by_id["brd"]["depends_on"], [])
        self.assertEqual(by_id["prd"]["depends_on"], ["brd"])
        self.assertEqual(by_id["sdd"]["depends_on"], ["prd"])
        self.assertEqual(by_id["benchmark"]["depends_on"], ["sdd"])

    def test_does_not_duplicate_dependency_already_present(self):
        manifest = {
            "nodes": [{"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}],
            "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        self.assertEqual(result["nodes"][1]["depends_on"], ["brd"])

    def test_edge_referencing_unknown_node_id_ignored(self):
        manifest = {"nodes": [{"id": "brd"}], "edges": [{"from": "brd", "to": "no-such-node"}]}
        result = work_graph_plan._normalize_graph_manifest(manifest)
        self.assertEqual(result["nodes"][0]["depends_on"], [])

    def test_zero_dependency_info_leaves_nodes_independent(self):
        # 2026-08-08 (FR-B4 "全部自由规划"): the canonical-doc-chain forced-
        # order enforcement that used to backfill this exact case (2026-08-03
        # incident) is deliberately removed, not extended -- Planner's node
        # set is no longer a fixed architectural constant. A real response
        # with zero dependency info now genuinely produces zero depends_on,
        # same as any other node set; robustness rests entirely on the
        # edge-derived backfill (test_depends_on_derived_from_edges_when_absent)
        # plus _validate_loop_back_targets, not a hardcoded chain to guess from.
        manifest = {
            "nodes": [
                {"node_id": "brd", "artifact_type": "brd"},
                {"node_id": "prd", "artifact_type": "prd"},
                {"node_id": "sdd", "artifact_type": "sdd"},
                {"node_id": "benchmark", "artifact_type": "benchmark"},
                {"node_id": "test_dataset", "artifact_type": "test_dataset"},
            ],
            "edges": [],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        for node in result["nodes"]:
            self.assertEqual(node["depends_on"], [])

    def test_on_failure_edge_does_not_get_folded_into_depends_on(self):
        # 2026-08-08: an on_failure edge is control-flow, not a data
        # dependency -- the edge-derived depends_on backfill above must
        # skip it (same reasoning as work_graph_orchestrate.py's
        # topological_layers()/_transitive_dependents).
        manifest = {
            "nodes": [
                {"node_id": "sdd", "artifact_type": "sdd"},
                {
                    "node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"],
                    "on_fail": {"action": "loop_back"},
                },
            ],
            "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}],
        }
        result = work_graph_plan._normalize_graph_manifest(manifest)
        by_id = {n["node_id"]: n for n in result["nodes"]}
        self.assertEqual(by_id["sdd"]["depends_on"], [])


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(work_graph_plan._transitive_ancestors(manifest, "c"), {"a", "b", "c"})
        self.assertEqual(work_graph_plan._transitive_ancestors(manifest, "a"), {"a"})

    def test_ignores_on_failure_edges(self):
        manifest = {
            "nodes": [{"node_id": "sdd"}, {"node_id": "benchmark"}],
            "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}],
        }
        # an on_failure edge alone (no real depends_on/data edge) must NOT
        # make benchmark an ancestor of sdd.
        self.assertEqual(work_graph_plan._transitive_ancestors(manifest, "sdd"), {"sdd"})


class ValidateLoopBackTargetsTest(unittest.TestCase):
    def test_valid_loop_back_passes(self):
        manifest = {
            "nodes": [
                {"node_id": "sdd", "depends_on": []},
                {
                    "node_id": "benchmark", "depends_on": ["sdd"],
                    "on_fail": {"action": "loop_back"},
                },
            ],
            "edges": [{"from": "benchmark", "to": "sdd", "edge_type": "on_failure"}],
        }
        work_graph_plan._validate_loop_back_targets(manifest)  # must not raise

    def test_no_on_failure_edge_raises(self):
        manifest = {
            "nodes": [
                {"node_id": "sdd", "depends_on": []},
                {"node_id": "benchmark", "depends_on": ["sdd"], "on_fail": {"action": "loop_back"}},
            ],
            "edges": [],
        }
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._validate_loop_back_targets(manifest)

    def test_two_on_failure_edges_from_same_node_raises(self):
        manifest = {
            "nodes": [
                {"node_id": "sdd", "depends_on": []},
                {"node_id": "prd", "depends_on": []},
                {
                    "node_id": "benchmark", "depends_on": ["sdd", "prd"],
                    "on_fail": {"action": "loop_back"},
                },
            ],
            "edges": [
                {"from": "benchmark", "to": "sdd", "edge_type": "on_failure"},
                {"from": "benchmark", "to": "prd", "edge_type": "on_failure"},
            ],
        }
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._validate_loop_back_targets(manifest)

    def test_target_not_an_upstream_dependency_raises(self):
        # sdd and unrelated are siblings with no dependency relationship --
        # benchmark depends on sdd but its on_failure edge points at the
        # unrelated node instead, which never influenced benchmark at all.
        manifest = {
            "nodes": [
                {"node_id": "sdd", "depends_on": []},
                {"node_id": "unrelated", "depends_on": []},
                {
                    "node_id": "benchmark", "depends_on": ["sdd"],
                    "on_fail": {"action": "loop_back"},
                },
            ],
            "edges": [{"from": "benchmark", "to": "unrelated", "edge_type": "on_failure"}],
        }
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._validate_loop_back_targets(manifest)

    def test_target_not_in_manifest_at_all_raises(self):
        manifest = {
            "nodes": [{"node_id": "benchmark", "depends_on": [], "on_fail": {"action": "loop_back"}}],
            "edges": [{"from": "benchmark", "to": "nonexistent", "edge_type": "on_failure"}],
        }
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._validate_loop_back_targets(manifest)

    def test_nodes_without_loop_back_are_unaffected(self):
        manifest = {
            "nodes": [
                {"node_id": "brd", "depends_on": []},
                {"node_id": "prd", "depends_on": ["brd"]},
            ],
            "edges": [],
        }
        work_graph_plan._validate_loop_back_targets(manifest)  # must not raise

    def test_normalize_graph_manifest_calls_validation_and_raises_on_bad_loop_back(self):
        manifest = {
            "nodes": [{"node_id": "benchmark", "on_fail": {"action": "loop_back"}}],
            "edges": [],
        }
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._normalize_graph_manifest(manifest)

    def test_target_in_additional_valid_target_ids_passes(self):
        # Real incident (2026-08-13, task b243dfe8): append mode validates a
        # newly-planned manifest BEFORE merging it with the already-persisted
        # one -- a new node legitimately looping back to an earlier,
        # already-persisted node (from a prior append) must not be rejected
        # just because that node isn't part of THIS manifest yet.
        manifest = {
            "nodes": [
                {"node_id": "test_script_verify", "depends_on": [], "on_fail": {"action": "loop_back"}},
            ],
            "edges": [{"from": "test_script_verify", "to": "sdd", "edge_type": "on_failure"}],
        }
        work_graph_plan._validate_loop_back_targets(
            manifest, additional_valid_target_ids={"sdd"}
        )  # must not raise
        with self.assertRaises(work_graph_plan.PlanningFailedError):
            work_graph_plan._validate_loop_back_targets(manifest)  # without it, still rejected


class HappyPathTest(PlanWorkGraphTestBase):
    def test_returns_run_id_and_graph_manifest(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            result = work_graph_plan.plan_work_graph(
                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
            )
        self.assertEqual(result["graph_manifest"], _MANIFEST)
        self.assertTrue(result["run_id"])

    def test_creates_work_graph_run_row_in_planning_status(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            result = work_graph_plan.plan_work_graph(
                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
            )
        row = self.conn.execute(
            "SELECT task_id, status, graph_manifest FROM work_graph_run WHERE run_id = ?",
            (result["run_id"],),
        ).fetchone()
        self.assertEqual(row[0], "task-1")
        self.assertEqual(row[1], "planning")
        self.assertEqual(json.loads(row[2]), _MANIFEST)

    def test_invokes_claude_with_expected_args(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")

        args, kwargs = run.call_args
        argv = args[0]
        self.assertEqual(argv[:3], ["claude", "--agent", "planner"])
        self.assertIn("--permission-mode", argv)
        self.assertIn("auto", argv)
        self.assertNotIn("--bare", argv)
        # --json-schema deliberately dropped -- confirmed for real it's a no-op
        # when combined with --agent (see module docstring)
        self.assertNotIn("--json-schema", argv)
        self.assertEqual(kwargs["cwd"], "/tmp/proj-1")
        self.assertEqual(kwargs["stdin"], subprocess.DEVNULL)


class FailureTest(PlanWorkGraphTestBase):
    def test_nonzero_exit_raises_planning_failed(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(returncode=1, stdout_obj=None, stderr="boom")
            with self.assertRaises(work_graph_plan.PlanningFailedError):
                work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")

    def test_is_error_true_raises_planning_failed(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(stdout_obj={"is_error": True})
            with self.assertRaises(work_graph_plan.PlanningFailedError):
                work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")

    def test_unparseable_result_raises_planning_failed(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(stdout_obj={"is_error": False, "result": "no json here"})
            with self.assertRaises(work_graph_plan.PlanningFailedError):
                work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")
        self.assertEqual(self.conn.execute("SELECT COUNT(*) FROM work_graph_run").fetchone()[0], 0)


# E (2026-08-06): cross-Org-Agent dynamic planning path -- org_node != 'planner'
# gets _GENERIC_PROMPT_TEMPLATE instead of the fixed doc-chain prompt, and each
# node carries its own task_prompt instead of a fixed artifact_type.

_CODER_THREE_MODULE_MANIFEST = {
    "nodes": [
        {"node_id": "test-module-a", "depends_on": [], "task_prompt": "Write unit tests for module_a.py."},
        {"node_id": "test-module-b", "depends_on": [], "task_prompt": "Write unit tests for module_b.py."},
        {"node_id": "test-module-c", "depends_on": [], "task_prompt": "Write unit tests for module_c.py."},
    ],
    "edges": [],
}


class GenericPlanningPromptTest(PlanWorkGraphTestBase):
    def test_planner_gets_the_self_planning_catalog_template(self):
        # 2026-08-08 (FR-B4 "全部自由规划"): Planner no longer gets a fixed
        # "produce exactly BRD/PRD/SDD/Benchmark" instruction -- it gets the
        # node type catalog and decides itself which nodes this goal needs.
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "goal", "/tmp/proj-1")
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertIn("Available node types", prompt)
        self.assertIn("\"artifact_type\": \"benchmark\"", prompt)
        self.assertIn("on_failure", prompt)
        self.assertNotIn("design-dynamic-work-graph", prompt)
        self.assertNotIn("{goal_text}", prompt)  # the placeholder itself must have been substituted

    def test_coder_gets_generic_template_referencing_the_skill(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertIn("design-dynamic-work-graph", prompt)
        self.assertNotIn("BRD, PRD, SDD, Benchmark", prompt)
        self.assertIn("task_prompt", prompt)

    def test_coder_prompt_includes_its_own_suggested_catalog_not_validators(self):
        # Phase 2 §3 (2026-08-12): each org_node gets its OWN suggested
        # node-type catalog, not a shared one -- "starting points, not
        # requirements", same contract _PLANNER_NODE_CATALOG already uses.
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertIn("Suggested node types for this goal", prompt)
        self.assertIn('"node_type": "unit_test"', prompt)
        self.assertIn('"node_type": "integration_check"', prompt)
        self.assertNotIn('"node_type": "independent_benchmark"', prompt)  # validator-only
        self.assertIn("promotes_to_artifact", prompt)
        self.assertIn("dynamic_loop_back", prompt)  # explicitly told NOT to use it

    def test_validator_prompt_includes_its_own_catalog_not_coders(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "validator", "task-1", "goal", "/tmp/proj-1")
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertIn('"node_type": "independent_benchmark"', prompt)
        self.assertIn('"node_type": "test_script"', prompt)
        self.assertNotIn('"node_type": "unit_test"', prompt)  # coder-only

    def test_unknown_org_node_gets_no_catalog_section_at_all(self):
        # An org_node this hasn't been extended for yet (Phase 1's
        # planner_reviewer, for instance) still gets the generic path --
        # just with no suggested catalog, same as before catalogs existed.
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "planner_reviewer", "task-1", "goal", "/tmp/proj-1")
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertNotIn("Suggested node types for this goal", prompt)
        self.assertIn("task_prompt", prompt)  # still the generic template itself


class AppendToRunIdTest(PlanWorkGraphTestBase):
    """Phase 2 §6.1 (2026-08-12): plan_work_graph(..., append_to_run_id=...)
    merges new nodes into an EXISTING work_graph_run instead of INSERTing a
    fresh row."""

    def _seed_run(self, run_id, manifest):
        self.conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) VALUES (?, 'task-1', ?, 'running')",
            (run_id, json.dumps(manifest)),
        )
        self.conn.commit()

    def test_merges_new_nodes_into_existing_run_not_a_new_row(self):
        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []}
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
            )
            result = work_graph_plan.plan_work_graph(
                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
            )

        self.assertEqual(result["run_id"], "run-1")
        self.assertEqual(result["new_node_ids"], ["impl"])
        merged_node_ids = {n["node_id"] for n in result["graph_manifest"]["nodes"]}
        self.assertEqual(merged_node_ids, {"sdd", "impl"})

        # exactly one work_graph_run row for this task -- no second row created.
        row_count = self.conn.execute("SELECT COUNT(*) FROM work_graph_run WHERE task_id = 'task-1'").fetchone()[0]
        self.assertEqual(row_count, 1)
        persisted = json.loads(
            self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
        )
        self.assertEqual({n["node_id"] for n in persisted["nodes"]}, {"sdd", "impl"})

    def test_existing_node_ids_are_injected_into_the_prompt(self):
        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": ["sdd"], "task_prompt": "do it"}], "edges": []}
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
            )
            work_graph_plan.plan_work_graph(
                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
            )
        prompt = run.call_args.args[0][run.call_args.args[0].index("-p") + 1]
        self.assertIn("already has these nodes", prompt)
        self.assertIn("sdd", prompt)
        self.assertIn("ADDITIONAL nodes only", prompt)

    def test_node_id_collision_with_existing_node_raises(self):
        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
        # the "new" plan reuses "sdd" as its own node_id -- a real collision.
        colliding_manifest = {"nodes": [{"node_id": "sdd", "depends_on": [], "task_prompt": "oops"}], "edges": []}
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(colliding_manifest)}
            )
            with self.assertRaises(work_graph_plan.PlanningFailedError):
                work_graph_plan.plan_work_graph(
                    self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
                )
        # the original run's manifest must be untouched by the failed append.
        persisted = json.loads(
            self.conn.execute("SELECT graph_manifest FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
        )
        self.assertEqual([n["node_id"] for n in persisted["nodes"]], ["sdd"])

    def test_new_node_loop_back_targeting_an_existing_node_is_allowed(self):
        # Real incident (2026-08-13, task b243dfe8): a genuinely valid
        # append -- a newly-planned verification node's on_failure edge
        # targets an already-persisted node from an earlier append. This
        # must NOT be rejected just because the earlier node isn't part of
        # the brand-new manifest by itself (it only exists once merged).
        self._seed_run("run-1", {"nodes": [{"node_id": "sdd", "artifact_type": "sdd", "depends_on": []}], "edges": []})
        new_manifest = {
            "nodes": [{
                "node_id": "verify", "depends_on": ["sdd"], "task_prompt": "verify it",
                "on_fail": {"action": "loop_back"},
            }],
            "edges": [{"from": "verify", "to": "sdd", "edge_type": "on_failure"}],
        }
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
            )
            result = work_graph_plan.plan_work_graph(  # must not raise
                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="run-1",
            )
        merged_node_ids = {n["node_id"] for n in result["graph_manifest"]["nodes"]}
        self.assertEqual(merged_node_ids, {"sdd", "verify"})

    def test_append_to_nonexistent_run_id_raises(self):
        new_manifest = {"nodes": [{"node_id": "impl", "depends_on": [], "task_prompt": "do it"}], "edges": []}
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(new_manifest)}
            )
            with self.assertRaises(ValueError):
                work_graph_plan.plan_work_graph(
                    self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id="no-such-run",
                )


class CoderThreeIndependentModulesScenarioTest(PlanWorkGraphTestBase):
    """E acceptance criterion 1: a Coder scenario that naturally splits into
    2-3 independent sub-tasks produces a graph_manifest via the new generic
    path (not the old single trigger_org_node call), correctly marking those
    nodes as mutually independent."""

    def test_three_independent_modules_produce_correctly_shaped_manifest(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            result = work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")

        manifest = result["graph_manifest"]
        self.assertEqual(len(manifest["nodes"]), 3)
        for node in manifest["nodes"]:
            self.assertEqual(node["depends_on"], [])
            self.assertIn("task_prompt", node)

        # E's whole point ties directly into G: independence is expressed
        # structurally (empty depends_on), so topological_layers() must put
        # all three in ONE concurrent layer, not three serial ones.
        layers = work_graph_orchestrate.topological_layers(manifest)
        self.assertEqual(len(layers), 1)
        self.assertEqual({n["node_id"] for n in layers[0]}, {"test-module-a", "test-module-b", "test-module-c"})


class GenericNodePromptBuilderTest(unittest.TestCase):
    def test_returns_the_node_own_task_prompt(self):
        node = {"node_id": "test-module-a", "task_prompt": "Write unit tests for module_a.py."}
        self.assertEqual(
            work_graph_plan.generic_node_prompt_builder(node, {}), "Write unit tests for module_a.py."
        )


class GenericPathEndToEndViaRunGraphTest(PlanWorkGraphTestBase):
    """E acceptance criterion (FR-5): the generated graph_manifest, driven by
    generic_node_prompt_builder, actually runs to completion through
    run_graph() -- not just a standalone parsing/shape check."""

    def test_three_independent_nodes_execute_and_return_outputs(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_CODER_THREE_MODULE_MANIFEST)}
            )
            plan_result = work_graph_plan.plan_work_graph(self.conn, "coder", "task-1", "goal", "/tmp/proj-1")

        # plan_work_graph() already inserted the work_graph_run row
        # (status='planning') -- run_graph() doesn't require any particular
        # status on it, so nothing more to set up here.
        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: {"content": "done: " + prompt}
            )
            outputs = work_graph_orchestrate.run_graph(
                self.conn, plan_result["run_id"], "coder", "coder-implementation", "/tmp/proj-1",
                plan_result["graph_manifest"], work_graph_plan.generic_node_prompt_builder,
                {"type": "object", "required": ["node_id"]}, {"type": "object", "required": ["content"]},
            )

        self.assertEqual(set(outputs.keys()), {"test-module-a", "test-module-b", "test-module-c"})
        self.assertEqual(mock_trigger.call_count, 3)


class PlanningNodeRecordedTest(PlanWorkGraphTestBase):
    """The planning call itself (this class's own `run.return_value`
    subprocess call) must show up as a real, clickable work_node_execution
    -- not just as the graph_manifest it produced -- so v_task_detail's
    to-do tree/DAG have a first node for "how was this run planned",
    per the user's own request (2026-08-15, task b243dfe8 discussion)."""

    def test_planning_node_execution_row_is_real_and_done(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            result = work_graph_plan.plan_work_graph(
                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
            )
        row = self.conn.execute(
            "SELECT run_id, subagent_type, skill_ref, status, output_payload "
            "FROM work_node_execution WHERE node_id = ?",
            (work_graph_plan.PLANNING_NODE_ID,),
        ).fetchone()
        self.assertIsNotNone(row)
        self.assertEqual(row[0], result["run_id"])
        self.assertEqual(row[1], "planner")
        self.assertEqual(row[2], "design-dynamic-work-graph")
        self.assertEqual(row[3], "done")
        self.assertEqual(json.loads(row[4])["graph_manifest"], _MANIFEST)

    def test_persisted_work_graph_run_manifest_stays_pure_for_execution(self):
        # run_graph()/fork/append-mode all read work_graph_run.graph_manifest
        # straight off the DB -- it must never gain the synthetic planning
        # node, or a later append_to_run_id call / a fork would try to treat
        # "planning itself" as executable work. See _record_planning_node's
        # own docstring.
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            result = work_graph_plan.plan_work_graph(
                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
            )
        persisted = self.conn.execute(
            "SELECT graph_manifest FROM work_graph_run WHERE run_id = ?", (result["run_id"],)
        ).fetchone()[0]
        self.assertEqual(json.loads(persisted), _MANIFEST)

    def test_log_viewer_content_includes_prompt_and_graph_manifest(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1")
        node_exec_id = self.conn.execute(
            "SELECT node_exec_id FROM work_node_execution WHERE node_id = ?",
            (work_graph_plan.PLANNING_NODE_ID,),
        ).fetchone()[0]
        log = execution_log.get_log(self.conn, node_exec_id)
        self.assertIn("node_start", log["content"])
        self.assertIn("node_completed", log["content"])
        self.assertIn("实现贪吃蛇游戏", log["content"])
        self.assertIn("n1", log["content"])  # a real node_id from _MANIFEST, proving the manifest is in there

    def test_task_query_prepends_planning_node_as_the_first_node(self):
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            work_graph_plan.plan_work_graph(self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1")

        task = task_query.get_task(self.conn, "task-1")
        display_nodes = task["work_graph_runs"][0]["graph_manifest"]["nodes"]
        self.assertEqual(display_nodes[0]["node_id"], work_graph_plan.PLANNING_NODE_ID)
        self.assertEqual([n["node_id"] for n in display_nodes[1:]], [n["node_id"] for n in _MANIFEST["nodes"]])
        # the synthetic node is clickable exactly like any other -- it has a
        # matching work_node_execution row in "nodes" with a real node_exec_id.
        wne_by_id = {n["node_id"]: n for n in task["work_graph_runs"][0]["nodes"]}
        self.assertIn(work_graph_plan.PLANNING_NODE_ID, wne_by_id)
        self.assertTrue(wne_by_id[work_graph_plan.PLANNING_NODE_ID]["node_exec_id"])

    def test_append_mode_does_not_record_a_second_planning_node(self):
        # YAGNI (see _record_planning_node's own docstring): only the fresh
        # branch gets one. Confirms append mode doesn't crash or duplicate.
        with patch("src.work_graph_plan.subprocess.run") as run:
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(_MANIFEST)}
            )
            first = work_graph_plan.plan_work_graph(
                self.conn, "planner", "task-1", "实现贪吃蛇游戏", "/tmp/proj-1"
            )
            run.return_value = _completed_process(
                stdout_obj={"is_error": False, "result": _fenced_result(
                    {"nodes": [{"node_id": "n5", "artifact_type": "test_dataset", "depends_on": []}], "edges": []}
                )}
            )
            work_graph_plan.plan_work_graph(
                self.conn, "coder", "task-1", "goal", "/tmp/proj-1", append_to_run_id=first["run_id"],
            )
        count = self.conn.execute(
            "SELECT COUNT(*) FROM work_node_execution WHERE run_id = ? AND node_id = ?",
            (first["run_id"], work_graph_plan.PLANNING_NODE_ID),
        ).fetchone()[0]
        self.assertEqual(count, 1)


if __name__ == "__main__":
    unittest.main()