"""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 json
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

_FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"}

_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"]},
    ],
    "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"},
    ],
}


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_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_four_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"]},
            ]
            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"})
        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"))


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"]},
                {"response": "Implemented. Ran `pytest`: 3 passed."},
            ]
            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"})
        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]
        self.assertEqual(all_artifacts, 6)  # brd, prd, sdd, benchmark, code_diff, test_report

        # planner and coder 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"})


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 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")
        mock_ensure.assert_called_once()
        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")


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