宜博 / 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="")


# 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": [],
}


_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 both Planner's and Coder's planning calls (which
    now share the same mocked subprocess.run patch target) each get their
    own realistic fake manifest instead of one hardcoded response serving
    both callers (which would hand Coder Planner's own brd/prd/... nodes,
    colliding with the ones already in the run it's 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`, 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]
    manifest = _FAKE_CODER_GRAPH_MANIFEST if org_node == "coder" else _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, org_node, graph_manifest, status) "
            "VALUES ('run-1', 'task-1', 'coder', '{}', '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(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_dispatch), \
             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"]},
                # 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."},
                # 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)

        # Phase 2 §6 (2026-08-12): Coder no longer gets its own work_graph_run
        # -- implement_code() appends its self-planned nodes onto Planner's
        # SAME run (plan_work_graph(..., append_to_run_id=...)), so only TWO
        # rows exist now, not three: the shared Planner+Coder run (still
        # labeled org_node='planner' -- that column reflects who ORIGINATED
        # the run, not every org_node that has since contributed nodes to it;
        # Phase 2c is expected to fold Validator into the same run too, at
        # which point this column stops being a reliable "who's involved"
        # signal at all, see design doc §2.1) and Validator's own separate run.
        run_rows = self.conn.execute(
            "SELECT run_id, org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],)
        ).fetchall()
        self.assertEqual(len(run_rows), 2)
        self.assertEqual({row[1] for row in run_rows}, {"planner", "validator"})

        # Coder's self-planned "impl" node genuinely landed in the SAME run
        # as Planner's nodes, not a separate one.
        planner_run_id = next(row[0] for row in run_rows if row[1] == "planner")
        node_ids = {
            row[0] for row in self.conn.execute(
                "SELECT node_id FROM work_node_execution WHERE run_id = ?", (planner_run_id,)
            ).fetchall()
        }
        self.assertIn("impl", 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, org_node, graph_manifest, status) "
                "VALUES ('run-1', ?, 'planner', '{}', '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)


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