宜博 / lemongraph

"""U-014 tests: end-to-end orchestration wiring, mocked at the subprocess
boundary (CLAUDE.md Testing constraint) -- these test that plan_documents/
implement_code/validate_implementation/run_end_to_end call the right things
in the right order and wire artifact lineage/task-status transitions
correctly, not what a real Planner/Coder/Validator would actually produce.
Real coverage (as much of it as is safe/affordable to run) is in
src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1).

**Rewired 2026-08-01 (U-031 regression audit)**: plan_documents/
implement_code now go through work_graph_plan.plan_work_graph +
work_graph_orchestrate.run_graph (which calls work_graph_execute.
execute_work_graph per node) instead of a flat org_node_trigger.
trigger_org_node call -- see e2e_pipeline.py's module docstring for why.
plan_work_graph is mocked at its own subprocess boundary (same pattern as
test_work_graph_plan.py) so its real graph_manifest-parsing logic and its
real work_graph_run INSERT both still run for real; execute_work_graph is
mocked directly (one level higher than its own subprocess boundary) since
this file's job is testing e2e_pipeline's wiring, not re-proving
execute_work_graph's own mechanics (covered in test_work_graph_execute.py)
or work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py).

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

import contextlib
import json
import os
import subprocess
import tempfile
import unittest
from unittest.mock import patch

from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute, work_graph_orchestrate

_FAKE_DOCS = {
    "brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake",
    "test_dataset": "N/A: no natural dataset for this fake goal",
    "test_script": "assert True  # fake but real, compilable script",
}

# test_script disabled 2026-08-07 (docs/todo/validator-test-script-
# narrative-output-issue.md) -- _PROMPT_TEMPLATE no longer asks for it, so
# the realistic fake planning response matches that. Tests that still need
# to exercise a test_script node (e.g. the partial-failure salvage
# mechanism, which is generic and doesn't care whether test_script is
# currently enabled by default) build their own local manifest instead of
# using this shared fixture.
_FAKE_GRAPH_MANIFEST = {
    "nodes": [
        {"node_id": "brd", "artifact_type": "brd", "depends_on": []},
        {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]},
        {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]},
        {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]},
        {"node_id": "test_dataset", "artifact_type": "test_dataset", "depends_on": ["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"},
        {"from": "benchmark", "to": "test_dataset", "edge_type": "needs_judgment"},
    ],
}

# Only used by tests that need a test_script node present despite it being
# disabled by default (see _FAKE_GRAPH_MANIFEST's comment above).
_FAKE_GRAPH_MANIFEST_WITH_TEST_SCRIPT = {
    "nodes": _FAKE_GRAPH_MANIFEST["nodes"] + [
        # depends_on both -- _DOC_PROMPTS["test_script"] needs
        # {benchmark_content} directly, not just what survives into
        # test_dataset_content. work_graph_plan._normalize_graph_manifest
        # used to auto-inject this second dependency for the canonical
        # chain's test_script node; that logic was removed along with
        # test_script from _CANONICAL_DOC_CHAIN (2026-08-07), so this fixture
        # sets it explicitly instead of relying on any auto-injection.
        # 2026-08-08: `judgment` is required now -- plan_documents() no
        # longer passes an external verify_fn at all (Work Graph
        # self-planning + loop-back, FR-B1); a node only gets checked if it
        # declares its own judgment, same as a real self-planned node would
        # by reusing work_graph_plan._PLANNER_NODE_CATALOG's suggestion.
        {
            "node_id": "test_script", "artifact_type": "test_script",
            "depends_on": ["test_dataset", "benchmark"],
            "judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
            # max_attempts=1 -- "1 retry allowed" (matches the legacy exactly-
            # one-retry default): exactly 2 real executions before exceeding,
            # matching this fixture's tests providing exactly 2 mock responses.
            "on_fail": {"action": "retry_self", "max_attempts": 1},
        },
    ],
    "edges": _FAKE_GRAPH_MANIFEST["edges"] + [
        {"from": "test_dataset", "to": "test_script", "edge_type": "needs_judgment"},
        {"from": "benchmark", "to": "test_script", "edge_type": "needs_judgment"},
    ],
}


def _fake_plan_subprocess_run(*args, **kwargs):
    payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_GRAPH_MANIFEST) + "\n```"}
    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")


def _fake_plan_subprocess_run_with_test_script(*args, **kwargs):
    payload = {
        "is_error": False,
        "result": "```json\n" + json.dumps(_FAKE_GRAPH_MANIFEST_WITH_TEST_SCRIPT) + "\n```",
    }
    return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")


def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300):
    assert prompt == e2e_pipeline._READY_PROMPT
    return "Ready."


class PipelineTestBase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
        self.conn = schema.connect(":memory:")
        schema.create_schema(self.conn)
        project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)

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


class PlanDocumentsTest(PipelineTestBase):
    def test_produces_five_artifacts_with_correct_lineage(self):
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', (SELECT project_id FROM project), '实现贪吃蛇游戏', 'goal', "
            "'planning_brd_prd_sdd', 'planner')"
        )
        self.conn.commit()

        with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
             patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
            mock_execute.side_effect = [
                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                {"content": _FAKE_DOCS["test_dataset"]},
            ]
            plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name)

        artifact_ids = plan_result["artifact_ids"]
        self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark", "test_dataset"})
        prd_deps = json.loads(
            self.conn.execute(
                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["prd"],)
            ).fetchone()[0]
        )
        self.assertEqual(prd_deps, [artifact_ids["brd"]])
        sdd_deps = json.loads(
            self.conn.execute(
                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?", (artifact_ids["sdd"],)
            ).fetchone()[0]
        )
        self.assertEqual(sdd_deps, [artifact_ids["prd"]])

        # plan_work_graph's own real work_graph_run INSERT actually ran (not bypassed)
        run_row = self.conn.execute(
            "SELECT org_node, status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
        ).fetchone()
        self.assertEqual(run_row, ("planner", "planning"))

    def test_earlier_nodes_survive_a_later_node_failing_verification(self):
        # 2026-08-07 real incident (docs/todo/work-graph-partial-artifact-
        # persistence-requirements.md): test_script fails verify_fn twice
        # (original + one regeneration) -- ConsistencyCheckFailedError must
        # still propagate (failure semantics unchanged), but BRD/PRD/SDD/
        # Benchmark/TestDataset -- all genuinely produced and accepted --
        # must be persisted as artifacts, not silently discarded.
        #
        # test_script itself is disabled by default as of the same date
        # (docs/todo/validator-test-script-narrative-output-issue.md) -- this
        # test uses _fake_plan_subprocess_run_with_test_script to keep
        # exercising the salvage mechanism itself (generic, not specific to
        # test_script) rather than losing this coverage entirely.
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', "
            "'planning_brd_prd_sdd', 'planner')"
        )
        self.conn.commit()

        # _fake_plan_subprocess_run_with_test_script ignores its args and
        # always returns a canned success payload -- fine for the "claude"
        # planning call it's meant for, but `patch("src.work_graph_plan.
        # subprocess.run", ...)` patches the ONE global `subprocess` module
        # every other importer shares too (work_graph_plan does `import
        # subprocess`, not `from subprocess import run`), so it would ALSO
        # silently hijack check_script_compiles's real `py_compile`/`node
        # --check` calls, making them always report success regardless of
        # actual content -- exactly defeating the failure this test needs to
        # trigger. Only intercept the "claude" planning call; let everything
        # else (including this test's real compile checks) run for real.
        real_subprocess_run = subprocess.run

        def _discriminating_fake(*args, **kwargs):
            cmd = args[0] if args else kwargs.get("args")
            if cmd and cmd[0] == "claude":
                return _fake_plan_subprocess_run_with_test_script(*args, **kwargs)
            return real_subprocess_run(*args, **kwargs)

        with patch("src.work_graph_plan.subprocess.run", side_effect=_discriminating_fake), \
             patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
            mock_execute.side_effect = [
                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                {"content": _FAKE_DOCS["test_dataset"]},
                {"content": "this is prose, not a real script"},  # test_script: fails compile check
                {"content": "still not real code either"},  # the one regeneration attempt: also fails
            ]
            with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
                e2e_pipeline.plan_documents(self.conn, "task-1", "goal", self.tmp.name)

        rows = self.conn.execute("SELECT artifact_type FROM artifact WHERE task_id = 'task-1'").fetchall()
        persisted = {r[0] for r in rows}
        self.assertEqual(persisted, {"brd", "prd", "sdd", "benchmark", "test_dataset"})
        self.assertNotIn("test_script", persisted)


class RunEndToEndTest(PipelineTestBase):
    def test_full_pipeline_wiring(self):
        with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \
             patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \
             patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."):
            mock_execute.side_effect = [
                {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                {"content": _FAKE_DOCS["test_dataset"]},
                {"response": "Implemented. Ran `pytest`: 3 passed."},
                # Validator's OWN independent benchmark/dataset (2026-08-06) --
                # deliberately different content from the Planner's, to also prove
                # nothing here accidentally reuses the Planner's copies.
                {"content": "# Validator's own Benchmark\nfake"},
                {"content": "N/A: validator judged no dataset needed either"},
            ]
            result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        self.assertEqual(
            set(result["doc_artifact_ids"]), {"brd", "prd", "sdd", "benchmark", "test_dataset"}
        )
        self.assertEqual(set(result["code_artifact_ids"]), {"code_diff", "test_report"})

        task_row = self.conn.execute(
            "SELECT status FROM task WHERE task_id = ?", (result["task_id"],)
        ).fetchone()
        self.assertEqual(task_row[0], "validating")

        all_artifacts = self.conn.execute(
            "SELECT COUNT(*) FROM artifact WHERE task_id = ?", (result["task_id"],)
        ).fetchone()[0]
        # brd, prd, sdd, benchmark, test_dataset, code_diff, test_report,
        # validator_benchmark, validator_test_dataset
        # (test_script/validator_test_script disabled, see _VALIDATOR_MANIFEST)
        self.assertEqual(all_artifacts, 9)

        # planner, coder, AND validator (2026-08-06: validator now has its own
        # real work_graph_run too, not a flat direct call) each got their own
        # work_graph_run (not sharing one)
        run_org_nodes = {
            row[0] for row in self.conn.execute(
                "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
            ).fetchall()
        }
        self.assertEqual(run_org_nodes, {"planner", "coder", "validator"})


class PrdConsistencyVerifyFnTest(unittest.TestCase):
    """REQ-02 boundary: only the PRD node gets the consistency check."""

    def test_non_prd_node_passes_through_without_a_call(self):
        node_artifact_types = {"brd": "brd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
            ok, feedback = verify_fn({"node_id": "brd"}, {"content": "brd text"})
        self.assertTrue(ok)
        self.assertIsNone(feedback)
        mock_execute.assert_not_called()

    def test_string_false_from_a_real_call_is_not_treated_as_truthy(self):
        # regression: a real call (2026-08-02) returned "consistent": "false"
        # as a JSON string, not a boolean -- `if "false":` is truthy in
        # Python, silently inverting the check until a real test caught it.
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": "false", "issues": ["contradiction found"]},
             ):
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
        self.assertFalse(ok)

    def test_string_true_from_a_real_call_is_treated_as_consistent(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": "true", "issues": []},
             ):
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "consistent prd"})
        self.assertTrue(ok)

    def test_prd_node_triggers_a_real_check_call(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": False, "issues": ["contradiction found"]},
             ) as mock_execute:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
        self.assertFalse(ok)
        self.assertEqual(feedback, ["contradiction found"])
        mock_execute.assert_called_once()
        args = mock_execute.call_args[0]
        self.assertEqual(args[0], "/tmp/proj-1")


class CheckScriptCompilesTest(unittest.TestCase):
    """2026-08-06: the one mechanical (not LLM-judged) verify_fn -- real,
    unmocked subprocess calls to py_compile/node --check, no fixtures to
    fake since this is testing an actual compiler's real opinion."""

    def test_na_content_passes_without_compiling(self):
        ok, detail = e2e_pipeline.check_script_compiles("N/A: this goal has no automatable check")
        self.assertTrue(ok)
        self.assertIsNone(detail)

    def test_valid_python_passes(self):
        ok, detail = e2e_pipeline.check_script_compiles("def test_thing():\n    assert 1 + 1 == 2\n")
        self.assertTrue(ok)

    def test_valid_javascript_passes(self):
        ok, detail = e2e_pipeline.check_script_compiles("function testThing() { return 1 + 1; }\n")
        self.assertTrue(ok)

    def test_narrative_prose_fails_both(self):
        # this is the exact failure mode that motivated this check -- a
        # narrative sentence about a script, not a real script.
        ok, detail = e2e_pipeline.check_script_compiles(
            "The graphlemon planner Work Node workflow completed successfully and produced "
            "a test script covering 18 scenarios across four areas."
        )
        self.assertFalse(ok)
        self.assertIn("py_compile", detail)
        self.assertIn("node --check", detail)

    def test_syntactically_broken_python_and_js_fails(self):
        ok, detail = e2e_pipeline.check_script_compiles("def broken(:\n    this is not valid anything\n")
        self.assertFalse(ok)


class TestScriptVerifyFnTest(unittest.TestCase):
    def test_non_test_script_node_passes_through_without_compiling(self):
        node_artifact_types = {"brd": "brd"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "brd"}, {"content": "not even close to code {{{"})
        self.assertTrue(ok)
        self.assertIsNone(detail)

    def test_test_script_node_with_broken_content_fails(self):
        node_artifact_types = {"script": "test_script"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "script"}, {"content": "this is narrative, not code, no way"})
        self.assertFalse(ok)
        # regression: work_graph_orchestrate does "; ".join(feedback or []) on
        # this return value -- a bare string would get joined character-by-
        # character into a garbled reflect prompt (2026-08-06 self-hosting
        # incident). Must be a list containing the message, not the message.
        self.assertIsInstance(detail, list)
        self.assertIn("py_compile", detail[0])

    def test_test_script_node_with_real_code_passes(self):
        node_artifact_types = {"script": "test_script"}
        verify_fn = e2e_pipeline._test_script_verify_fn(node_artifact_types)
        ok, detail = verify_fn({"node_id": "script"}, {"content": "assert True\n"})
        self.assertTrue(ok)


class PlannerPromptBuilderTest(unittest.TestCase):
    """2026-08-08: found via a real self-planned run whose SDD node had no
    PRD node at all (a legitimate Work Graph self-planning choice, FR-B4
    "全部自由规划") -- _DOC_PROMPTS["sdd"] hardcodes a {prd_content}
    placeholder left over from the old fixed chain, and plain .format()
    raised a bare KeyError, landing the task on blocked_pending_human with
    an unhelpful generic 'execution_error' reason instead of a real result."""

    def test_missing_upstream_dependency_content_does_not_raise(self):
        node_artifact_types = {"sdd": "sdd"}
        build_prompt = e2e_pipeline._planner_prompt_builder("fix a bug", node_artifact_types)
        # sdd's depends_on is empty -- no prd node was ever planned, so
        # {prd_content} has nothing to fill it. Must not raise KeyError.
        prompt = build_prompt({"node_id": "sdd", "depends_on": []}, {})
        self.assertIn("N/A (no corresponding upstream node was included in this Work Graph)", prompt)
        self.assertIn("fix a bug", prompt)

    def test_present_upstream_dependency_content_is_used_normally(self):
        node_artifact_types = {"prd": "prd", "sdd": "sdd"}
        build_prompt = e2e_pipeline._planner_prompt_builder("goal", node_artifact_types)
        prompt = build_prompt(
            {"node_id": "sdd", "depends_on": ["prd"]}, {"prd": {"content": "real prd text"}}
        )
        self.assertIn("real prd text", prompt)
        self.assertNotIn("N/A (no corresponding", prompt)


class CombinedPlannerVerifyFnTest(unittest.TestCase):
    """Dispatches to the right per-node-type check -- both real checks
    (PRD consistency and script-compiles) are exercised elsewhere in
    isolation; this only tests the dispatch itself."""

    def test_prd_node_dispatches_to_prd_check(self):
        node_artifact_types = {"n1": "prd"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            ok, detail = verify_fn({"node_id": "n1"}, {"content": "a prd"})
        self.assertTrue(ok)
        mock_execute.assert_called_once()

    def test_test_script_node_dispatches_to_compile_check(self):
        node_artifact_types = {"n1": "test_script"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        ok, detail = verify_fn({"node_id": "n1"}, {"content": "not code at all, just words"})
        self.assertFalse(ok)

    def test_other_node_types_pass_through(self):
        node_artifact_types = {"n1": "test_dataset"}
        verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1")
        ok, detail = verify_fn({"node_id": "n1"}, {"content": "anything at all"})
        self.assertTrue(ok)
        self.assertIsNone(detail)


class PrdConsistencySelfTargetWorktreeTest(unittest.TestCase):
    """2026-08-03 regression: a real self-hosting run showed _check_prd_
    consistency() calling execute_work_graph() directly against
    target_folder_path, bypassing worktree isolation entirely -- when
    self-targeting, its own agentType='planner'+memory:project call writes
    a real, uncommitted change straight to the main checkout's
    .claude/agent-memory/planner/MEMORY.md, which then made the NEXT
    node's own (correctly worktree-isolated) merge fail with "local
    changes would be overwritten by merge". This call must isolate the
    same way work_node_run.run_work_node() already does for every other
    self-target call."""

    def test_self_target_isolates_the_check_in_its_own_worktree(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/repo")
        with patch.object(e2e_pipeline.paths, "is_self_target", return_value=True), \
             patch.object(e2e_pipeline.work_node_run, "create_worktree", return_value="/repo/.graphlemon-worktrees/x") as mock_create, \
             patch.object(e2e_pipeline.work_node_run, "ensure_worktree_committed") as mock_ensure, \
             patch.object(e2e_pipeline.work_node_run, "merge_worktree") as mock_merge, \
             patch.object(e2e_pipeline.work_node_run, "remove_worktree") as mock_remove, \
             patch.object(work_graph_execute, "write_workflow_script") as mock_write, \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "some prd"})

        self.assertTrue(ok)
        mock_create.assert_called_once_with("/repo", unittest.mock.ANY)
        # both the script write and the real execute call must use the
        # WORKTREE path as cwd, never the main checkout, when self-targeting
        self.assertEqual(mock_write.call_args[0][0], "/repo/.graphlemon-worktrees/x")
        self.assertEqual(mock_execute.call_args[0][0], "/repo/.graphlemon-worktrees/x")
        # called twice: once for the worktree itself, once for the main
        # checkout (worktrees share project identity, so a real memory
        # write can land on the main checkout regardless of cwd -- see
        # work_node_run.ensure_worktree_committed's own docstring)
        self.assertEqual(mock_ensure.call_count, 2)
        ensure_paths = [call.args[0] for call in mock_ensure.call_args_list]
        self.assertIn("/repo/.graphlemon-worktrees/x", ensure_paths)
        self.assertIn("/repo", ensure_paths)
        mock_merge.assert_called_once()
        mock_remove.assert_called_once()

    def test_non_self_target_does_not_use_a_worktree_at_all(self):
        node_artifact_types = {"prd": "prd"}
        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
        with patch.object(e2e_pipeline.paths, "is_self_target", return_value=False), \
             patch.object(e2e_pipeline.work_node_run, "create_worktree") as mock_create, \
             patch.object(work_graph_execute, "write_workflow_script") as mock_write, \
             patch.object(
                 work_graph_execute, "execute_work_graph",
                 return_value={"consistent": True, "issues": []},
             ) as mock_execute:
            verify_fn({"node_id": "prd"}, {"content": "some prd"})

        mock_create.assert_not_called()
        self.assertEqual(mock_write.call_args[0][0], "/tmp/proj-1")
        self.assertEqual(mock_execute.call_args[0][0], "/tmp/proj-1")


class ValidateImplementationTest(unittest.TestCase):
    def test_parses_pass_verdict(self):
        with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: PASS\nlooks good"):
            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
        self.assertEqual(result["verdict"], "PASS")

    def test_parses_fail_verdict(self):
        with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: FAIL\nwall collision broken"):
            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
        self.assertEqual(result["verdict"], "FAIL")

    def test_no_verdict_line_defaults_to_inconclusive(self):
        with patch("src.org_node_trigger.trigger_org_node", return_value="I'm not sure what happened."):
            result = e2e_pipeline.validate_implementation("task-1", "benchmark content", "goal", "/tmp/p")
        self.assertEqual(result["verdict"], "INCONCLUSIVE")


class RunTestScriptForRealTest(unittest.TestCase):
    """2026-08-06: real, unmocked execution -- these are testing what an
    actual interpreter actually does, same reasoning as CheckScriptCompilesTest."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)

    def test_na_content_is_not_executed(self):
        result = e2e_pipeline.run_test_script_for_real(
            "N/A: no automatable check for this goal", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertFalse(result["ran"])
        self.assertIsNone(result["exit_code"])

    def test_real_passing_python_script(self):
        result = e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(0)\n", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertTrue(result["ran"])
        self.assertEqual(result["exit_code"], 0)

    def test_real_failing_python_script(self):
        result = e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(1)\n", self.tmp.name, "task-1", "validator_test_script"
        )
        self.assertTrue(result["ran"])
        self.assertEqual(result["exit_code"], 1)

    def test_script_runs_with_real_cwd_so_it_sees_the_real_repo(self):
        with open(os.path.join(self.tmp.name, "marker.txt"), "w") as f:
            f.write("real file")
        script = (
            "import os, sys\n"
            "sys.exit(0 if os.path.exists('marker.txt') else 1)\n"
        )
        result = e2e_pipeline.run_test_script_for_real(script, self.tmp.name, "task-1", "validator_test_script")
        self.assertEqual(result["exit_code"], 0)

    def test_materializes_a_real_file_under_graphlemon_runs(self):
        e2e_pipeline.run_test_script_for_real(
            "import sys\nsys.exit(0)\n", self.tmp.name, "task-42", "validator_test_script"
        )
        expected = os.path.join(self.tmp.name, ".graphlemon", "runs", "task-42", "validator_test_script.py")
        self.assertTrue(os.path.exists(expected))


# validator_test_script disabled by default 2026-08-07 (see
# e2e_pipeline._VALIDATOR_MANIFEST's own comment) -- tests below that need
# to exercise the hard-gate/salvage mechanisms involving that node patch
# _VALIDATOR_MANIFEST back to this 3-node shape for their own duration, so
# that real code (still in the codebase, just not planned by default) stays
# covered instead of silently rotting.
_VALIDATOR_MANIFEST_WITH_TEST_SCRIPT = {
    "nodes": [
        {"node_id": "validator_benchmark", "artifact_type": "validator_benchmark", "depends_on": []},
        {
            "node_id": "validator_test_dataset", "artifact_type": "validator_test_dataset",
            "depends_on": ["validator_benchmark"],
        },
        {
            "node_id": "validator_test_script", "artifact_type": "validator_test_script",
            "depends_on": ["validator_test_dataset", "validator_benchmark"],
        },
    ],
    "edges": [],
}


class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
    """2026-08-06: the safety-critical piece -- a real non-zero exit from
    Validator's own script must force FAIL no matter what the LLM's verdict
    line says (user-confirmed hard gate)."""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.addCleanup(self.tmp.cleanup)
        subprocess.run(["git", "init"], cwd=self.tmp.name, check=True, capture_output=True)
        self.conn = schema.connect(":memory:")
        schema.create_schema(self.conn)
        project_scaffold.create_project(self.conn, self.tmp.name, init_git_if_missing=False)
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES ('task-1', (SELECT project_id FROM project), 'goal', 'goal', 'validating', 'validator')"
        )
        self.conn.commit()

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

    def _run(self, script_content, verdict_line, with_test_script=False):
        mock_responses = [
            {"content": "# Validator's own Benchmark"},
            {"content": "N/A: no dataset needed"},
        ]
        if with_test_script:
            mock_responses.append({"content": script_content})
        with contextlib.ExitStack() as stack:
            stack.enter_context(patch.object(work_graph_execute, "write_workflow_script"))
            mock_execute = stack.enter_context(patch.object(work_graph_execute, "execute_work_graph"))
            stack.enter_context(patch.object(org_node_trigger, "trigger_org_node", return_value=verdict_line))
            if with_test_script:
                stack.enter_context(
                    patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT)
                )
            mock_execute.side_effect = mock_responses
            return e2e_pipeline.validate_implementation_with_own_graph(
                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
            )

    def test_llm_says_pass_and_script_really_passes_stays_pass(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True)
        self.assertEqual(result["verdict"], "PASS")
        self.assertTrue(result["script_execution"]["ran"])
        self.assertEqual(result["script_execution"]["exit_code"], 0)

    def test_llm_says_pass_but_script_really_fails_is_force_overridden_to_fail(self):
        # the exact case the user's hard-gate rule exists for: a script
        # that objectively fails must not be argued away by the LLM's own
        # verdict line.
        result = self._run(
            "import sys\nsys.exit(1)\n", "VERDICT: PASS\nlooks good to me", with_test_script=True
        )
        self.assertEqual(result["verdict"], "FAIL")

    def test_llm_says_fail_and_script_fails_stays_fail(self):
        result = self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")
        self.assertEqual(result["verdict"], "FAIL")

    def test_na_script_leaves_llm_judgment_untouched(self):
        result = self._run("N/A: no mechanical check possible", "VERDICT: INCONCLUSIVE\nneeds human review")
        self.assertEqual(result["verdict"], "INCONCLUSIVE")
        self.assertFalse(result["script_execution"]["ran"])

    def test_writes_two_validator_artifacts_with_lineage(self):
        # test_script disabled by default -- see _VALIDATOR_MANIFEST_WITH_
        # TEST_SCRIPT variant below for the 3-artifact/script-lineage case.
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
        self.assertEqual(set(result["artifact_ids"]), {"validator_benchmark", "validator_test_dataset"})
        dataset_deps = json.loads(
            self.conn.execute(
                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
                (result["artifact_ids"]["validator_test_dataset"],),
            ).fetchone()[0]
        )
        self.assertIn(result["artifact_ids"]["validator_benchmark"], dataset_deps)

    def test_writes_three_validator_artifacts_with_lineage_when_test_script_re_enabled(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good", with_test_script=True)
        self.assertEqual(
            set(result["artifact_ids"]),
            {"validator_benchmark", "validator_test_dataset", "validator_test_script"},
        )
        script_deps = json.loads(
            self.conn.execute(
                "SELECT depends_on_artifact_ids FROM artifact WHERE artifact_id = ?",
                (result["artifact_ids"]["validator_test_script"],),
            ).fetchone()[0]
        )
        self.assertIn(result["artifact_ids"]["validator_benchmark"], script_deps)
        self.assertIn(result["artifact_ids"]["validator_test_dataset"], script_deps)

    def test_creates_its_own_work_graph_run_row(self):
        result = self._run("import sys\nsys.exit(0)\n", "VERDICT: PASS\nlooks good")
        run_row = self.conn.execute(
            "SELECT org_node, status FROM work_graph_run WHERE task_id = 'task-1'"
        ).fetchone()
        self.assertEqual(run_row[0], "validator")
        # 2026-08-07 (real incident): this used to stay 'running' forever on
        # the happy path -- verify_fn returning ok=True means
        # _verify_and_maybe_regenerate never touches work_graph_run.status
        # at all, so nothing else in this function's old code ever marked
        # it terminal. schema.py's `ux_work_graph_run_active` unique index
        # covers exactly this non-terminal status set, so a stuck 'running'
        # row silently blocked any later re-verify of the SAME task from
        # ever creating its own new validator run.
        self.assertEqual(run_row[1], "done")

    def test_second_validation_run_does_not_collide_with_the_first(self):
        # The actual regression: org_level_loop.run_fix_loop re-verifying
        # after a FAIL (or a reconciliation-loop resume retrying) calls this
        # function again for the SAME task_id -- must not raise
        # "UNIQUE constraint failed: work_graph_run.task_id,
        # work_graph_run.org_node" against a first run stuck non-terminal.
        self._run("import sys\nsys.exit(1)\n", "VERDICT: FAIL\nreal bug found")

        # Benchmark/dataset content must differ from the first call's --
        # write_artifact does a real `git commit`, which fails with "nothing
        # to commit" if a second write produces byte-identical file content
        # (the file path isn't run_id-scoped). Real re-verify runs would
        # naturally regenerate slightly different content each time; this
        # just makes that explicit rather than accidentally relying on it.
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nfixed now"):
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark (re-verify)"},
                {"content": "N/A: no dataset needed, still"},
                {"content": "import sys\nsys.exit(0)\n"},
            ]
            second = e2e_pipeline.validate_implementation_with_own_graph(
                self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
            )

        self.assertEqual(second["verdict"], "PASS")
        rows = self.conn.execute(
            "SELECT status FROM work_graph_run WHERE task_id = 'task-1' ORDER BY rowid"
        ).fetchall()
        self.assertEqual([r[0] for r in rows], ["done", "done"])

    def test_earlier_nodes_survive_a_later_node_failing_verification(self):
        # 2026-08-07 (same real incident/fix as PlanDocumentsTest's own
        # test of this name): validator_test_script fails verify_fn twice
        # -- ConsistencyCheckFailedError must still propagate, but
        # validator_benchmark/validator_test_dataset (genuinely produced
        # and accepted) must be persisted, not discarded along with it.
        # validator_test_script itself is disabled by default (see
        # _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT above) -- re-enabled here to
        # keep the salvage mechanism (generic, not test_script-specific) covered.
        with patch.object(work_graph_execute, "write_workflow_script"), \
             patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \
             patch.object(e2e_pipeline, "_VALIDATOR_MANIFEST", _VALIDATOR_MANIFEST_WITH_TEST_SCRIPT):
            mock_execute.side_effect = [
                {"content": "# Validator's own Benchmark"},
                {"content": "N/A: no dataset needed"},
                {"content": "this is prose, not a real script"},  # fails compile check
                {"content": "still not real code either"},  # the one regeneration attempt: also fails
            ]
            with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
                e2e_pipeline.validate_implementation_with_own_graph(
                    self.conn, "task-1", "sdd content", "prd content", "goal text", self.tmp.name,
                )

        rows = self.conn.execute("SELECT artifact_type FROM artifact WHERE task_id = 'task-1'").fetchall()
        persisted = {r[0] for r in rows}
        self.assertEqual(persisted, {"validator_benchmark", "validator_test_dataset"})
        self.assertNotIn("validator_test_script", persisted)
        # the work_graph_run itself must NOT be left stuck 'running' either
        # (ux_work_graph_run_active would block any later re-verify) --
        # _verify_and_maybe_regenerate's own 'blocked' write on this path
        # already covers it; confirming it here catches a regression if
        # that ever stops being true.
        run_status = self.conn.execute(
            "SELECT status FROM work_graph_run WHERE task_id = 'task-1'"
        ).fetchone()[0]
        self.assertEqual(run_status, "blocked")


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