宜博 / 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 org_node_trigger.
trigger_org_node_structured per node, 2026-08 deblackbox migration --
originally work_graph_execute.execute_work_graph, see that module's own
docstring) 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; trigger_org_node_structured 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 trigger_org_node_
structured's own mechanics (covered in test_org_node_trigger.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_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="")


# Phase 2 §6 (2026-08-12): Coder now self-plans too (implement_code() calls
# plan_work_graph(conn, "coder", ..., append_to_run_id=...), going through
# the SAME mocked subprocess boundary as Planner's own planning call) --
# depends_on Planner's sdd node so it lands in a LATER topological layer
# than Planner's already-'done' nodes, not the same one (both being
# in-degree-zero would trigger run_graph()'s concurrent ThreadPoolExecutor
# layer path purely as a mocking artifact, unrelated to what this test is
# actually checking).
_FAKE_CODER_GRAPH_MANIFEST = {
    "nodes": [{
        "node_id": "impl", "depends_on": ["sdd"], "task_prompt": "implement it",
        "promotes_to_artifact": {"artifact_type": "test_report"},
    }],
    "edges": [],
}

# Phase 2c (2026-08-12): Validator now self-plans too, into the SAME shared
# run as Planner/Coder -- depends_on "sdd" for the same "avoid landing in
# Planner's own already-'done' in-degree-zero layer" reason
# _FAKE_CODER_GRAPH_MANIFEST's own comment gives. A single node_type=
# 'final_verdict' node is the minimal real shape validate_implementation_
# with_own_graph() accepts (see work_graph_plan._VALIDATOR_NODE_CATALOG's
# own catalog entry) -- no test_script node here, this fixture is about
# pipeline wiring, not the hard-gate mechanism (see VerdictHardGateTest/
# ValidateImplementationWithOwnGraphTest above for that).
_FAKE_VALIDATOR_GRAPH_MANIFEST = {
    "nodes": [{
        "node_id": "verdict", "depends_on": ["sdd"], "task_prompt": "give your final verdict",
        "node_type": "final_verdict",
        "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
        "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
    }],
    "edges": [],
}


_real_subprocess_run = subprocess.run  # captured before any test patches subprocess.run globally


def _fake_plan_subprocess_run_dispatch(*args, **kwargs):
    """Real plan_work_graph() dispatches by the --agent argv value -- this
    mock does the same, so Planner's/Coder's/Validator's planning calls
    (which now all share the same mocked subprocess.run patch target) each
    get their own realistic fake manifest instead of one hardcoded response
    serving every caller (which would hand Coder/Validator Planner's own
    brd/prd/... nodes, colliding with the ones already in the run they're
    appending onto).

    patch("src.work_graph_plan.subprocess.run", ...) patches subprocess.run
    GLOBALLY, not just within work_graph_plan.py -- `subprocess` is the same
    singleton module object everywhere it's imported, so this side_effect
    also intercepts every OTHER real subprocess.run call this test's own
    call chain makes (artifact_write.write_artifact's `git add`/`git
    commit`, e2e_pipeline's `git diff`/`git rev-parse`, the final_verdict
    node's own REAL test-script execution, etc). Those aren't
    `claude --agent ... -p ...` calls -- pass them through to the REAL
    subprocess.run instead of returning a canned fake "claude planning
    response" CompletedProcess for them too (which happened to not visibly
    break anything before this dispatch function existed, only because
    nothing downstream read those specific calls' .stdout in a way that
    exposed it -- still wrong to keep doing on purpose now that this
    function actually inspects argv)."""
    argv = args[0]
    if not argv or argv[0] != "claude":
        return _real_subprocess_run(*args, **kwargs)
    org_node = argv[argv.index("--agent") + 1]
    if org_node == "coder":
        manifest = _FAKE_CODER_GRAPH_MANIFEST
    elif org_node == "validator":
        manifest = _FAKE_VALIDATOR_GRAPH_MANIFEST
    else:
        manifest = _FAKE_GRAPH_MANIFEST
    payload = {"is_error": False, "result": "```json\n" + json.dumps(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)
        # check_same_thread=False (2026-08-12, Phase 2 §6) -- see
        # test_task_execution.py's TaskExecutionTestBase for why: a merged
        # graph's concurrent-layer path can now be reached even in tests
        # that don't otherwise care about G's parallel execution.
        self.conn = schema.connect(":memory:", check_same_thread=False)
        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 WriteCompletedNodeArtifactsTest(PipelineTestBase):
    """Phase 2 §2.3/§6 (2026-08-12) direct coverage: promotes_to_artifact
    resolution and only_node_ids scoping, tested directly rather than only
    incidentally through plan_documents()/implement_code() integration
    tests."""

    def setUp(self):
        super().setUp()
        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', 'coding', 'coder')"
        )
        self.conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', '{}', 'running')"
        )
        self.conn.execute(
            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
            "VALUES ('coder-implementation', 'coder', 'skills/coder-implementation/', 'unknown', 'new', 'active')"
        )
        self.conn.commit()

    def _mark_done(self, node_id, content):
        self.conn.execute(
            "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
            "input_schema_ref, output_schema_ref, input_payload, output_payload, status) "
            "VALUES (?, 'run-1', ?, 'coder', 'coder-implementation', 'in.json', 'out.json', '{}', ?, 'done')",
            ("node-exec-" + node_id, node_id, json.dumps({"content": content})),
        )
        self.conn.commit()

    def test_node_with_no_artifact_type_or_promotes_to_artifact_is_skipped(self):
        # most self-planned Coder nodes are internal graph steps, not final
        # deliverables -- must be silently skipped, not an error.
        manifest = {"nodes": [{"node_id": "explore", "depends_on": []}], "edges": []}
        self._mark_done("explore", "some intermediate reasoning")
        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
            self.conn, self.tmp.name, "task-1", "run-1", manifest, {}, lambda t: "md",
        )
        self.assertEqual(artifact_ids, {})
        count = self.conn.execute("SELECT COUNT(*) FROM artifact WHERE task_id = 'task-1'").fetchone()[0]
        self.assertEqual(count, 0)

    def test_promotes_to_artifact_node_is_persisted(self):
        manifest = {
            "nodes": [{
                "node_id": "final_report", "depends_on": [],
                "promotes_to_artifact": {"artifact_type": "test_report"},
            }],
            "edges": [],
        }
        self._mark_done("final_report", "# Real report content")
        node_artifact_types = {"final_report": "test_report"}
        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
            self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
        )
        self.assertEqual(set(artifact_ids), {"test_report"})
        content = self.conn.execute(
            "SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_ids["test_report"],)
        ).fetchone()[0]
        with open(os.path.join(self.tmp.name, content)) as f:
            self.assertEqual(f.read(), "# Real report content")

    def test_only_node_ids_scopes_to_new_nodes_skipping_earlier_promotable_ones(self):
        # a merged graph (Planner's already-artifact'd nodes + Coder's just-
        # appended ones) must not re-promote/re-process nodes an EARLIER
        # stage already wrote artifacts for, even though they'd otherwise
        # be individually promotable.
        manifest = {
            "nodes": [
                {"node_id": "sdd", "artifact_type": "sdd", "depends_on": []},
                {
                    "node_id": "impl", "depends_on": ["sdd"],
                    "promotes_to_artifact": {"artifact_type": "test_report"},
                },
            ],
            "edges": [],
        }
        self._mark_done("sdd", "# SDD content")
        self._mark_done("impl", "# Report content")
        node_artifact_types = {"sdd": "sdd", "impl": "test_report"}
        artifact_ids = e2e_pipeline._write_completed_node_artifacts(
            self.conn, self.tmp.name, "task-1", "run-1", manifest, node_artifact_types, lambda t: "md",
            only_node_ids={"impl"},
        )
        self.assertEqual(set(artifact_ids), {"test_report"})  # NOT "sdd" -- outside only_node_ids


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(org_node_trigger, "trigger_org_node_structured") as mock_trigger:
            mock_trigger.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 status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],)
        ).fetchone()
        self.assertEqual(run_row, ("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/done/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(org_node_trigger, "trigger_org_node_structured") as mock_trigger:
            mock_trigger.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_dispatch), \
             patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger, \
             patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."):
            mock_trigger.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"]},
                # Phase 2 §6 (2026-08-12): Coder self-plans now -- one
                # self-planned "impl" node (_FAKE_CODER_GRAPH_MANIFEST),
                # content-keyed like every other self-planned node, not the
                # old single synthetic node's response-keyed output.
                {"content": "Implemented. Ran `pytest`: 3 passed."},
                # Phase 2c (2026-08-12): Validator's own verdict is now a
                # self-planned node too (_FAKE_VALIDATOR_GRAPH_MANIFEST),
                # not a flat org_node_trigger.trigger_org_node call.
                {"content": "VERDICT: PASS\nAll scenarios pass."},
            ]
            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 --
        # Phase 2c: Validator's self-planned "verdict" node deliberately
        # does NOT declare promotes_to_artifact (a verdict is a judgment,
        # not a persisted deliverable -- see _generic_prompt_template's own
        # instruction), so no validator_* artifacts exist anymore.
        self.assertEqual(all_artifacts, 7)

        # Phase 2c (2026-08-12): Validator now ALSO appends its self-planned
        # nodes onto the SAME shared run (plan_work_graph(..., append_to_
        # run_id=...)) instead of creating its own separate row -- only ONE
        # work_graph_run row exists for this task now. org_node column
        # removed entirely (§2.1) -- real per-node ownership lives in
        # work_node_execution.subagent_type, checked below via node_ids.
        run_rows = self.conn.execute(
            "SELECT run_id, status FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
        ).fetchall()
        self.assertEqual(len(run_rows), 1)
        self.assertEqual(run_rows[0][1], "done")

        # Coder's "impl" AND Validator's "verdict" nodes both genuinely
        # landed in the SAME run as Planner's own nodes.
        shared_run_id = run_rows[0][0]
        node_ids = {
            row[0] for row in self.conn.execute(
                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,)
            ).fetchall()
        }
        self.assertIn("impl", node_ids)
        self.assertIn("verdict", node_ids)
        self.assertIn("sdd", node_ids)


class LooksLikeNarrationTest(unittest.TestCase):
    """Real repro fixtures from docs/todo/validator-test-script-narrative-
    output-issue.md -- these exact phrasings recurred across multiple
    independent task runs."""

    def test_real_repro_phrases_flagged(self):
        self.assertTrue(e2e_pipeline.looks_like_narration(
            "Launched the 'graphlemon-node-f9cefb87af4c492aac738b9736d41e4e' workflow "
            "(Work Node validator) in the background, run ID wf_93345ca8-8c6. It's running "
            "asynchronously — I'll report results once it completes; no further action is "
            "needed right now."
        ))
        self.assertTrue(e2e_pipeline.looks_like_narration(
            "The graphlemon Work Node (validator) workflow completed. It authored "
            "test_nav_project_context_benchmark.js — an independent mechanical test."
        ))
        self.assertTrue(e2e_pipeline.looks_like_narration(
            "You'll be notified automatically when it completes; no further action is needed right now."
        ))

    def test_real_structured_content_not_flagged(self):
        self.assertFalse(e2e_pipeline.looks_like_narration(
            "# SDD\n\n## Modules\n\n- auth: handles login/logout\n\n## Data flow\n\n..."
        ))

    def test_empty_or_none_not_flagged(self):
        self.assertFalse(e2e_pipeline.looks_like_narration(""))
        self.assertFalse(e2e_pipeline.looks_like_narration(None))


class ReviewPlannerOutputTest(PipelineTestBase):
    def test_pass_verdict_parsed(self):
        with patch.object(
            org_node_trigger, "trigger_org_node",
            return_value="VERDICT: PASS\nStructured, consistent with goal.",
        ) as mock_trigger:
            result = e2e_pipeline.review_planner_output(
                self.conn, "task-1", "goal", self.tmp.name, sdd_content="# SDD\nreal content",
            )
        self.assertEqual(result["verdict"], "PASS")
        self.assertEqual(mock_trigger.call_args[0][0], "planner_reviewer")

    def test_fail_verdict_parsed(self):
        with patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: FAIL\nNarrative prose."):
            result = e2e_pipeline.review_planner_output(
                self.conn, "task-1", "goal", self.tmp.name, sdd_content="I launched the workflow.",
            )
        self.assertEqual(result["verdict"], "FAIL")

    def test_narration_heuristic_adds_hint_to_prompt(self):
        captured = {}

        def fake_trigger(org_node, prompt, target_folder_path, timeout=300):
            captured["prompt"] = prompt
            return "VERDICT: FAIL\nreal narration"

        with patch.object(org_node_trigger, "trigger_org_node", side_effect=fake_trigger):
            e2e_pipeline.review_planner_output(
                self.conn, "task-1", "goal", self.tmp.name,
                sdd_content="Launched the workflow in the background, no further action is needed.",
            )
        self.assertIn("Heuristic flag: SDD", captured["prompt"])

    def test_no_narration_no_hint_in_prompt(self):
        captured = {}

        def fake_trigger(org_node, prompt, target_folder_path, timeout=300):
            captured["prompt"] = prompt
            return "VERDICT: PASS"

        with patch.object(org_node_trigger, "trigger_org_node", side_effect=fake_trigger):
            e2e_pipeline.review_planner_output(
                self.conn, "task-1", "goal", self.tmp.name, sdd_content="# Real SDD content",
            )
        self.assertNotIn("Heuristic flag", captured["prompt"])


class ReviewPlannerOutputWithRetriesTest(PipelineTestBase):
    def setUp(self):
        super().setUp()
        self.task_id = "task-1"
        self.conn.execute(
            "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
            "VALUES (?, (SELECT project_id FROM project LIMIT 1), 'goal', 'goal', "
            "'reviewing_planner_output', 'planner_reviewer')",
            (self.task_id,),
        )
        self.conn.commit()

    def test_pass_on_first_attempt_reuses_existing_artifacts_no_replan(self):
        for artifact_type, content in (("sdd", "# SDD\nreal"), ("brd", "# BRD\nreal")):
            self.conn.execute(
                "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
                "VALUES ('run-1', ?, '{}', 'done') ON CONFLICT(run_id) DO NOTHING",
                (self.task_id,),
            )
            file_path = ".graphlemon/runs/{}/{}.md".format(self.task_id, artifact_type)
            os.makedirs(os.path.dirname(os.path.join(self.tmp.name, file_path)), exist_ok=True)
            with open(os.path.join(self.tmp.name, file_path), "w") as f:
                f.write(content)
            self.conn.execute(
                "INSERT INTO artifact (artifact_id, artifact_type, task_id, produced_by_run_id, "
                "file_path, git_commit_sha, validation_status) VALUES (?, ?, ?, 'run-1', ?, 'sha', 'passed')",
                (artifact_type + "-artifact", artifact_type, self.task_id, file_path),
            )
        self.conn.commit()

        with patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \
             patch.object(e2e_pipeline, "plan_documents") as mock_plan:
            result = e2e_pipeline._review_planner_output_with_retries(
                self.conn, self.task_id, "goal", self.tmp.name,
            )
        mock_plan.assert_not_called()
        self.assertEqual(result["artifact_ids"]["sdd"], "sdd-artifact")
        mock_trigger.assert_called_once()

    def test_zero_existing_artifacts_calls_plan_documents(self):
        # P0-bug-adjacent scenario: no doc artifact exists at all (e.g. the
        # Planner graph failed on its very first node) -- must fall through
        # to a real plan_documents() call, not assume something exists.
        with patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"), \
             patch.object(
                 e2e_pipeline, "plan_documents",
                 return_value={"run_id": "run-x", "artifact_ids": {"sdd": "sdd-artifact"}},
             ) as mock_plan, \
             patch.object(e2e_pipeline, "_read_artifact_content", return_value="# SDD\nfake"):
            result = e2e_pipeline._review_planner_output_with_retries(
                self.conn, self.task_id, "goal", self.tmp.name,
            )
        mock_plan.assert_called_once()
        self.assertEqual(result["artifact_ids"]["sdd"], "sdd-artifact")

    def test_exhausting_retries_raises_loop_limit_error(self):
        with patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: FAIL\nstill narrative"), \
             patch.object(
                 e2e_pipeline, "plan_documents",
                 return_value={"run_id": "run-x", "artifact_ids": {"sdd": "sdd-artifact"}},
             ) as mock_plan, \
             patch.object(e2e_pipeline, "_read_artifact_content", return_value="I launched the workflow."):
            with self.assertRaises(e2e_pipeline.PlannerReviewLoopLimitReachedError):
                e2e_pipeline._review_planner_output_with_retries(
                    self.conn, self.task_id, "goal", self.tmp.name,
                )
        self.assertEqual(mock_plan.call_count, e2e_pipeline._PLANNER_REVIEW_MAX_ATTEMPTS)

        events = self.conn.execute(
            "SELECT event_type FROM execution_event WHERE task_id = ? AND event_type = 'planner_review_verdict'",
            (self.task_id,),
        ).fetchall()
        self.assertEqual(len(events), e2e_pipeline._PLANNER_REVIEW_MAX_ATTEMPTS)

    def test_rejected_first_attempt_replans_into_same_run_not_a_second_row(self):
        """Real prod incident (2026-08-13): ux_work_graph_run_active allows
        only ONE non-terminal work_graph_run row per task_id, and a rejected
        planner_reviewer attempt's row is never marked 'done'/'failed' just
        because its content was rejected -- a retry that calls
        plan_documents() with no run_id to append into does a fresh INSERT
        and hits 'sqlite3.IntegrityError: UNIQUE constraint failed:
        work_graph_run.task_id'. This test goes through the REAL
        plan_work_graph()/work_graph_run INSERT path (only the `claude`
        subprocess boundary and trigger_org_node_structured are mocked,
        same pattern as PlanDocumentsTest) so the real UNIQUE index is what
        actually catches a regression here, not a mocked-away assumption.
        Before the fix: raises IntegrityError on the second attempt. After:
        both attempts land in the SAME row, and only the newly-planned node
        actually re-executes (the first attempt's node is already 'done'
        and must be skipped, not silently re-run)."""

        def fake_plan_subprocess_run(*args, **kwargs):
            # patch("src.work_graph_plan.subprocess.run", ...) patches the
            # GLOBAL subprocess.run (see _fake_plan_subprocess_run_dispatch's
            # own comment above) -- only intercept the "claude" planning
            # call, pass everything else (artifact_write's `git add`/`git
            # commit`, etc) through to the real subprocess.run.
            argv = args[0]
            if not argv or argv[0] != "claude":
                return _real_subprocess_run(*args, **kwargs)
            # work_graph_plan.plan_work_graph() injects a "...ADDITIONAL
            # nodes only" marker into the prompt in append mode (see its
            # own docstring) -- used here the same way a real re-planning
            # LLM call is instructed to behave: pick a DIFFERENT node_id
            # from the rejected attempt's, not redeclare it. Real content
            # differs per attempt too (a genuine re-plan), which is why
            # this can't just reuse the first attempt's node_id even if
            # append mode didn't already forbid the collision outright.
            prompt = argv[argv.index("-p") + 1]
            node_id = "sdd_v2" if "ADDITIONAL nodes only" in prompt else "sdd"
            manifest = {"nodes": [{"node_id": node_id, "artifact_type": "sdd", "depends_on": []}], "edges": []}
            payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
            return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")

        with patch("src.work_graph_plan.subprocess.run", side_effect=fake_plan_subprocess_run), \
             patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger_structured, \
             patch.object(org_node_trigger, "trigger_org_node") as mock_verdict:
            mock_trigger_structured.side_effect = [
                {"content": "# SDD v1 -- rejected"},
                {"content": "# SDD v2 -- fixed"},
            ]
            mock_verdict.side_effect = ["VERDICT: FAIL\nnot structured enough", "VERDICT: PASS\nlooks real now"]

            result = e2e_pipeline._review_planner_output_with_retries(
                self.conn, self.task_id, "goal", self.tmp.name,
            )

        # Exactly one non-terminal work_graph_run row for this task_id --
        # the whole point of ux_work_graph_run_active. Before the fix this
        # assertion is moot: the second plan_documents() call never even
        # completes, it raises IntegrityError first.
        run_rows = self.conn.execute(
            "SELECT run_id, status FROM work_graph_run WHERE task_id = ?", (self.task_id,)
        ).fetchall()
        self.assertEqual(len(run_rows), 1)
        self.assertEqual(run_rows[0][1], "planning")
        shared_run_id = run_rows[0][0]
        self.assertEqual(result["run_id"], shared_run_id)

        # Both attempts' nodes exist in the SAME run's work_node_execution,
        # and the first attempt's already-'done' node was skipped (not
        # re-executed) the second time around -- exactly 2 trigger_org_
        # node_structured calls total, not 3.
        node_ids = {
            row[0] for row in self.conn.execute(
                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (shared_run_id,)
            ).fetchall()
        }
        self.assertEqual(node_ids, {"sdd", "sdd_v2"})
        self.assertEqual(mock_trigger_structured.call_count, 2)

        # The final PASS'd artifact is the SECOND attempt's content.
        sdd_content = e2e_pipeline._read_artifact_content(
            self.conn, self.tmp.name, result["artifact_ids"]["sdd"]
        )
        self.assertEqual(sdd_content, "# SDD v2 -- fixed")

        events = self.conn.execute(
            "SELECT event_type FROM execution_event WHERE task_id = ? AND event_type = 'planner_review_verdict'",
            (self.task_id,),
        ).fetchall()
        self.assertEqual(len(events), 2)


class RunTaskLifecycleMissingArtifactTypesTest(PipelineTestBase):
    """2026-08-08 regression (Org Graph unification Phase 1 §7): a
    self-planned graph_manifest omitting prd/benchmark used to crash
    run_task_lifecycle() with a bare KeyError on artifact_ids["prd"]/
    ["benchmark"] -- not resume-specific, this hit fresh first runs too."""

    def test_prd_and_benchmark_missing_does_not_raise_key_error(self):
        captured = {}

        def fake_implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, **kwargs):
            captured["benchmark_content"] = kwargs.get("benchmark_content")
            return {"code_diff": "diff-1", "test_report": "report-1"}

        def fake_validate(conn, task_id, run_id, sdd_content, prd_content, goal_text, target_folder_path, **kwargs):
            captured["prd_content"] = prd_content
            return {"verdict": "PASS", "raw_response": "ok"}

        with patch.object(
            e2e_pipeline, "plan_and_review_documents",
            return_value={"run_id": "run-1", "artifact_ids": {"sdd": "sdd-artifact"}},
        ), \
             patch.object(e2e_pipeline, "_read_artifact_content", return_value="# SDD\nreal"), \
             patch.object(e2e_pipeline.org_node_lifecycle, "trigger_org_node_full", return_value={}), \
             patch.object(e2e_pipeline, "implement_code", side_effect=fake_implement_code), \
             patch.object(e2e_pipeline, "validate_implementation_with_own_graph", side_effect=fake_validate):
            result = e2e_pipeline.run_task_lifecycle(self.conn, "task-1", self.tmp.name, "goal")

        self.assertEqual(result["verdict"]["verdict"], "PASS")
        self.assertEqual(captured["benchmark_content"], "N/A: not provided")
        self.assertEqual(captured["prd_content"], "N/A: not provided")


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(org_node_trigger, "trigger_org_node_structured") as mock_trigger:
            ok, feedback = verify_fn({"node_id": "brd"}, {"content": "brd text"})
        self.assertTrue(ok)
        self.assertIsNone(feedback)
        mock_trigger.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(
            org_node_trigger, "trigger_org_node_structured",
            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(
            org_node_trigger, "trigger_org_node_structured",
            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(
            org_node_trigger, "trigger_org_node_structured",
            return_value={"consistent": False, "issues": ["contradiction found"]},
        ) as mock_trigger:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
        self.assertFalse(ok)
        self.assertEqual(feedback, ["contradiction found"])
        mock_trigger.assert_called_once()
        args = mock_trigger.call_args[0]
        self.assertEqual(args[0], "planner")
        self.assertEqual(args[2], "/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(
            org_node_trigger, "trigger_org_node_structured",
            return_value={"consistent": True, "issues": []},
        ) as mock_trigger:
            ok, detail = verify_fn({"node_id": "n1"}, {"content": "a prd"})
        self.assertTrue(ok)
        mock_trigger.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(
                 org_node_trigger, "trigger_org_node_structured",
                 return_value={"consistent": True, "issues": []},
             ) as mock_trigger:
            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "some prd"})

        self.assertTrue(ok)
        mock_create.assert_called_once_with("/repo", unittest.mock.ANY)
        # the real trigger call must use the WORKTREE path as cwd (3rd
        # positional arg: org_node, prompt, target_folder_path), never the
        # main checkout, when self-targeting
        self.assertEqual(mock_trigger.call_args[0][2], "/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(
                 org_node_trigger, "trigger_org_node_structured",
                 return_value={"consistent": True, "issues": []},
             ) as mock_trigger:
            verify_fn({"node_id": "prd"}, {"content": "some prd"})

        mock_create.assert_not_called()
        self.assertEqual(mock_trigger.call_args[0][2], "/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))


def _fake_validator_plan_subprocess_run(manifest):
    # Same real-passthrough-for-non-'claude'-argv requirement as this
    # file's other subprocess.run fakes (see _fake_plan_subprocess_run_
    # dispatch's own docstring) -- the final_verdict node's build_prompt
    # wrapper actually EXECUTES a real test script via a real subprocess.run
    # call of its own (run_test_script_for_real/_detect_script_language),
    # which would otherwise also get intercepted by this same globally-
    # patched target and handed the fake planning JSON instead of a real
    # exit code.
    def run(*args, **kwargs):
        argv = args[0] if args else kwargs.get("args")
        if not argv or argv[0] != "claude":
            return _real_subprocess_run(*args, **kwargs)
        payload = {"is_error": False, "result": "```json\n" + json.dumps(manifest) + "\n```"}
        return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="")
    return run


# on_fail.action='hard_fail' below (not the real catalog's suggested
# 'dynamic_loop_back') is a deliberate test-fixture choice, not something a
# real self-planned graph would use: it isolates the hard-gate mechanical
# check itself (this file's job) from the separate dynamic_loop_back/
# Reflection-review retry machinery, which already has its own dedicated
# coverage in test_work_graph_orchestrate.py (DynamicLoopBackContentErrorTest/
# DynamicLoopBackSkillDefectTest/TallyReflectionTest) -- mocking a 3-perspective
# real reflection jury here would be expensive and wouldn't test anything new.
_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY = {
    "nodes": [{
        "node_id": "verdict", "depends_on": [], "task_prompt": "give your final verdict",
        "node_type": "final_verdict",
        "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
        "on_fail": {"action": "hard_fail"},
    }],
    "edges": [],
}

_FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT = {
    "nodes": [
        {
            "node_id": "test_script", "depends_on": [], "task_prompt": "write a real test script",
            "node_type": "test_script",
            "judgment": {"kind": "mechanical", "check_ref": "check_script_compiles"},
        },
        {
            "node_id": "verdict", "depends_on": ["test_script"], "task_prompt": "give your final verdict",
            "node_type": "final_verdict",
            "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
            "on_fail": {"action": "hard_fail"},
        },
    ],
    "edges": [],
}


class VerdictHardGateTest(unittest.TestCase):
    """Phase 2c (2026-08-12): unit coverage of the two functions carrying
    the safety-critical hard gate (2026-08-06, user-confirmed: a real
    non-zero exit from Validator's own test script forces FAIL regardless
    of what the LLM's verdict line says) forward into the self-planned
    final_verdict node design -- direct/isolated rather than through the
    full self-planning + run_graph() + work_node_run machinery, which the
    integration tests below cover for the wiring, not this specific logic."""

    def test_real_script_failure_overrides_pass_verdict_text(self):
        shared_state = {"script_execution": {"ran": True, "exit_code": 1, "stdout": "", "stderr": "boom"}}
        ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good")
        self.assertFalse(ok)
        self.assertIn("exit_code=1", detail)

    def test_passing_script_and_pass_verdict_is_ok(self):
        shared_state = {"script_execution": {"ran": True, "exit_code": 0, "stdout": "", "stderr": ""}}
        ok, detail = e2e_pipeline._verdict_says_pass_check(shared_state)("VERDICT: PASS\nlooks good")
        self.assertTrue(ok)
        self.assertIsNone(detail)

    def test_no_script_and_pass_verdict_is_ok(self):
        ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: PASS\nlooks good")
        self.assertTrue(ok)

    def test_fail_verdict_is_not_ok_even_with_no_script(self):
        ok, detail = e2e_pipeline._verdict_says_pass_check({})("VERDICT: FAIL\nreal bug found")
        self.assertFalse(ok)
        self.assertIn("FAIL", detail)

    def test_inconclusive_verdict_is_not_ok(self):
        # 2026-08-12 (Phase 2c, design decision): INCONCLUSIVE is treated
        # the same as FAIL here, not specially accepted -- see
        # validate_implementation_with_own_graph's own docstring for why.
        ok, _ = e2e_pipeline._verdict_says_pass_check({})("VERDICT: INCONCLUSIVE\nneeds human review")
        self.assertFalse(ok)

    def test_build_prompt_passes_through_unchanged_for_non_verdict_nodes(self):
        shared_state = {}
        inner = lambda node, outputs: "inner prompt for " + node["node_id"]
        wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60)
        result = wrapped({"node_id": "impl"}, {})
        self.assertEqual(result, "inner prompt for impl")
        self.assertEqual(shared_state, {})

    def test_build_prompt_runs_real_script_for_verdict_node_with_script_dependency(self):
        shared_state = {}
        inner = lambda node, outputs: "base prompt"
        node_types = {"test_script": "test_script"}
        outputs = {"test_script": {"content": "import sys\nsys.exit(0)\n"}}
        with tempfile.TemporaryDirectory() as tmp:
            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
            wrapped = e2e_pipeline._verdict_node_build_prompt(inner, node_types, "task-1", tmp, shared_state, 60)
            result = wrapped({"node_id": "verdict", "node_type": "final_verdict"}, outputs)
        self.assertTrue(shared_state["script_execution"]["ran"])
        self.assertEqual(shared_state["script_execution"]["exit_code"], 0)
        self.assertIn("base prompt", result)
        self.assertIn("REAL execution result", result)

    def test_build_prompt_na_stub_for_verdict_node_with_no_script_dependency(self):
        shared_state = {}
        inner = lambda node, outputs: "base prompt"
        wrapped = e2e_pipeline._verdict_node_build_prompt(inner, {}, "task-1", "/tmp", shared_state, 60)
        wrapped({"node_id": "verdict", "node_type": "final_verdict"}, {})
        self.assertFalse(shared_state["script_execution"]["ran"])


class ValidateImplementationWithOwnGraphTest(unittest.TestCase):
    """Phase 2c (2026-08-12): Validator now self-plans its own verification
    graph, appended onto the SAME shared run Planner/Coder already
    contributed to (plan_work_graph(..., append_to_run_id=run_id)) -- see
    e2e_pipeline.validate_implementation_with_own_graph's own docstring.
    VerdictHardGateTest above covers the hard-gate logic itself in
    isolation; these tests cover the wiring: planning dispatch, mechanical
    check registration, run_id reuse, and result unwrapping."""

    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:", check_same_thread=False)
        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')"
        )
        # The shared Planner+Coder run Validator appends onto -- empty on
        # purpose (these tests aren't about cross-stage node dependencies,
        # see test_task_execution.py for that), just needs to genuinely
        # exist for append_to_run_id's own existence check.
        self.conn.execute(
            "INSERT INTO work_graph_run (run_id, task_id, graph_manifest, status) "
            "VALUES ('run-shared', 'task-1', '{\"nodes\": [], \"edges\": []}', 'running')"
        )
        self.conn.commit()

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

    def _run(self, manifest, verdict_content, script_content=None):
        mock_responses = []
        if script_content is not None:
            mock_responses.append({"content": script_content})
        mock_responses.append({"content": verdict_content})
        with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_validator_plan_subprocess_run(manifest)), \
             patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger:
            mock_trigger.side_effect = mock_responses
            return e2e_pipeline.validate_implementation_with_own_graph(
                self.conn, "task-1", "run-shared", "sdd content", "prd content", "goal text", self.tmp.name,
            )

    def test_llm_says_pass_and_script_really_passes_stays_pass(self):
        result = self._run(
            _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good",
            script_content="import sys\nsys.exit(0)\n",
        )
        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_raises(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 -- hard_fail (see the fixture's own comment) makes
        # this deterministic without invoking the loop-back/reflection jury.
        with self.assertRaises(work_graph_orchestrate.ConsistencyCheckFailedError):
            self._run(
                _FAKE_VALIDATOR_MANIFEST_WITH_SCRIPT, "VERDICT: PASS\nlooks good to me",
                script_content="import sys\nsys.exit(1)\n",
            )

    def test_no_script_node_leaves_llm_judgment_as_the_only_signal(self):
        result = self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
        self.assertEqual(result["verdict"], "PASS")
        self.assertFalse(result["script_execution"]["ran"])

    def test_appends_into_the_shared_run_without_creating_a_new_row(self):
        self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
        rows = self.conn.execute("SELECT run_id, status FROM work_graph_run WHERE task_id = 'task-1'").fetchall()
        self.assertEqual(rows, [("run-shared", "done")])
        node_ids = {
            r[0] for r in self.conn.execute(
                "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'"
            ).fetchall()
        }
        self.assertIn("verdict", node_ids)

    def test_second_call_appends_fresh_nodes_without_colliding(self):
        # A real re-verify plans its OWN new node_ids each time (a fresh
        # LLM planning call, not a replay) -- node_id collision is what
        # plan_work_graph's append mode actively guards against (raises
        # PlanningFailedError), so this uses a different node_id for the
        # second call, same as a real second planning call naturally would.
        second_manifest = {
            "nodes": [{
                "node_id": "verdict_v2", "depends_on": [], "task_prompt": "give your final verdict",
                "node_type": "final_verdict",
                "judgment": {"kind": "mechanical", "check_ref": "verdict_says_pass"},
                "on_fail": {"action": "hard_fail"},
            }],
            "edges": [],
        }
        self._run(_FAKE_VALIDATOR_MANIFEST_VERDICT_ONLY, "VERDICT: PASS\nlooks good")
        second = self._run(second_manifest, "VERDICT: PASS\nfixed now")

        self.assertEqual(second["verdict"], "PASS")
        node_ids = {
            r[0] for r in self.conn.execute(
                "SELECT node_id FROM work_node_execution WHERE run_id = 'run-shared'"
            ).fetchall()
        }
        self.assertEqual(node_ids, {"verdict", "verdict_v2"})


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